feat: add `headroom perf` CLI and rewrite `headroom learn` to use LLM analysis

Proxy performance logging (`headroom perf`):
- Add always-on RotatingFileHandler to ~/.headroom/logs/proxy.log (10MB x 5 backups)
- Replace scattered log lines with structured PERF lines containing model, msgs,
tok_before/after/saved, cache_read/write/hit_pct, opt_ms, and transforms
- Emit PERF lines from all three response paths (streaming Anthropic, non-streaming
Anthropic, Bedrock streaming)
- Add `headroom perf` CLI that parses proxy logs and reports token savings, cache
hit rates, prefix stability, transform effectiveness, routing breakdown, TOIN
status, and actionable recommendations
- Support --hours and --raw flags for time filtering and raw record output

Learn module rewrite (LLM-based analysis):
- Replace all regex/heuristic analyzers with a single LLM call via LiteLLM
- New SessionAnalyzer builds compact digests and sends to any of 100+ models
- Auto-detect best model from API keys (Anthropic → OpenAI → Gemini)
- Add --model flag for explicit model selection
- Enrich scanner with SessionEvent (user messages, interruptions, subagent summaries),
token usage tracking, and timestamps
- Simplify models: remove EnvironmentFact, StructureNote, Correction, CommandPattern,
RetryPattern, AnalysisReport; add SessionEvent, AnalysisResult
- Simplify writer: remove Recommender class (LLM now produces recommendations directly)
- Update tests for new analyzer and models

headroom/cli/learn.py

@@ -89,32 +89,47 @@ def _auto_detect_agents() -> list[tuple[str, ConversationScanner, ContextWriter]
     default="auto",
     help=_AGENT_HELP,
 )
+@click.option(
+    "--model",
+    type=str,
+    default=None,
+    help="LLM model for analysis (e.g., claude-sonnet-4-6, gpt-4o, gemini/gemini-2.0-flash). "
+    "Auto-detected from API keys if not specified.",
+)
 def learn(
     project: Path | None,
     analyze_all: bool,
     apply: bool,
     agent: str,
+    model: str | None,
 ) -> None:
     """Learn from past tool call failures to prevent future ones.
 
-    Analyzes conversation history to find failure patterns (wrong paths,
-    missing modules, stubborn retries) and generates context that prevents
-    them from recurring.
+    Analyzes conversation history using an LLM to find failure patterns
+    (wrong paths, missing modules, stubborn retries) and generates context
+    that prevents them from recurring.
 
     Supports multiple coding agents: Claude Code, Codex, Gemini CLI.
-    Auto-detects which agents have data on your machine by default.
+    Uses LiteLLM for provider-agnostic LLM access (100+ models).
 
     \b
     Examples:
-        headroom learn                        # Auto-detect agent, analyze current project
+        headroom learn                        # Auto-detect agent & model
         headroom learn --apply                # Write recommendations
+        headroom learn --model gpt-4o         # Use GPT-4o for analysis
         headroom learn --all                  # Analyze all projects
         headroom learn --agent codex --all    # Analyze all Codex sessions
-        headroom learn --agent claude --project ~/myapp
     """
-    from ..learn.analyzer import FailureAnalyzer
+    from ..learn.analyzer import SessionAnalyzer, _detect_default_model
 
-    analyzer = FailureAnalyzer()
+    # Resolve model early to fail fast with a clear message
+    try:
+        resolved_model = model or _detect_default_model()
+    except RuntimeError as e:
+        click.echo(f"Error: {e}")
+        raise SystemExit(1) from None
+
+    analyzer = SessionAnalyzer(model=resolved_model)
 
     # Determine which agents to scan
     if agent == "auto":
@@ -171,25 +186,22 @@ def learn(
                 click.echo("  No conversation data found.")
                 continue
 
-            from ..learn.writer import Recommender
-
-            recommender = Recommender()
-            report = analyzer.analyze(proj, sessions)
+            click.echo(f"  Analyzing with {resolved_model}...")
+            result_data = analyzer.analyze(proj, sessions)
             total_projects += 1
-            total_failures += report.total_failures
+            total_failures += result_data.total_failures
 
             click.echo(
-                f"\n  Sessions: {report.total_sessions}  |  "
-                f"Calls: {report.total_calls}  |  "
-                f"Failures: {report.total_failures} ({report.failure_rate:.1%})  |  "
-                f"Corrections: {len(report.corrections)}"
+                f"\n  Sessions: {result_data.total_sessions}  |  "
+                f"Calls: {result_data.total_calls}  |  "
+                f"Failures: {result_data.total_failures} ({result_data.failure_rate:.1%})"
             )
 
-            if report.failure_rate == 0:
-                click.echo("  No failures found.")
+            if result_data.failure_rate == 0 and not result_data.recommendations:
+                click.echo("  No failures or patterns found.")
                 continue
 
-            recommendations = recommender.recommend(report)
+            recommendations = result_data.recommendations
             if not recommendations:
                 click.echo("  No actionable patterns found.")
                 continue

headroom/cli/main.py

@@ -38,6 +38,7 @@ def _register_commands() -> None:
         learn,  # noqa: F401
         mcp,  # noqa: F401
         memory,  # noqa: F401
+        perf,  # noqa: F401
         proxy,  # noqa: F401
     )
 

headroom/cli/perf.py

@@ -0,0 +1,48 @@
+"""Performance analysis CLI command."""
+
+import click
+
+from .main import main
+
+
+@main.command()
+@click.option(
+    "--hours",
+    type=float,
+    default=168.0,
+    help="Analyze logs from the last N hours (default: 168 = 7 days)",
+)
+@click.option("--raw", is_flag=True, help="Show raw PERF records instead of report")
+def perf(hours: float, raw: bool) -> None:
+    """Analyze proxy performance from logs.
+
+    \b
+    Reads logs from ~/.headroom/logs/proxy.log and shows:
+    - Token savings and compression effectiveness
+    - Cache hit rates and prefix stability
+    - Transform and routing breakdown
+    - TOIN learning status
+    - Actionable recommendations
+
+    \b
+    Examples:
+        headroom perf                Analyze last 7 days
+        headroom perf --hours 24     Analyze last 24 hours
+        headroom perf --raw          Show raw parsed records
+    """
+    from headroom.perf.analyzer import format_report, parse_log_files
+
+    report = parse_log_files(last_n_hours=hours)
+
+    if raw:
+        for r in report.perf_records:
+            click.echo(
+                f"{r.timestamp} {r.request_id} model={r.model} msgs={r.num_messages} "
+                f"before={r.tokens_before} after={r.tokens_after} saved={r.tokens_saved} "
+                f"cache_read={r.cache_read} cache_write={r.cache_write} "
+                f"cache_hit={r.cache_hit_pct}% opt={r.optimization_ms:.0f}ms"
+            )
+        if not report.perf_records:
+            click.echo("No PERF records found. Run the proxy first: headroom proxy")
+    else:
+        click.echo(format_report(report))

headroom/learn/__init__.py

@@ -1,17 +1,12 @@
-"""Headroom Learn — offline failure learning for coding agents.
+"""Headroom Learn — offline session learning for coding agents.
 
-Analyzes conversation logs to find tool call failure patterns and generates
-context (CLAUDE.md, MEMORY.md, .cursorrules, etc.) that prevents future failures.
+Analyzes conversation logs using Sonnet 4.6 to extract actionable patterns
+and generates context (CLAUDE.md, MEMORY.md, AGENTS.md, etc.) that prevents
+future token waste.
 
 Architecture:
-    Scanner (adapter)  →  Analyzer (generic)  →  Writer (adapter)
-    ├── ClaudeCodeScanner   ├── EnvironmentAnalyzer   ├── ClaudeCodeWriter
-    ├── CursorScanner       ├── StructureAnalyzer     ├── CursorWriter
-    └── GenericScanner      ├── RetryAnalyzer         └── GenericWriter
-                            ├── PermissionAnalyzer
-                            └── CrossSessionAnalyzer
-
-Scanners read tool-specific log formats and produce normalized ToolCall sequences.
-Analyzers work on ToolCall — same analysis for any agent system.
-Writers output to tool-specific context injection mechanisms.
+    Scanner (adapter)  →  Analyzer (LLM)  →  Writer (adapter)
+    ├── ClaudeCodeScanner   SessionAnalyzer     ├── ClaudeCodeWriter
+    └── CodexScanner        (Sonnet 4.6)        ├── CodexWriter
+                                                └── GeminiWriter
 """

headroom/learn/analyzer.py

@@ -1,640 +1,390 @@
-"""Failure analyzers with success correlation.
+"""Session analysis via LLM — replaces all regex/heuristic analysis.
 
-The core insight: don't just catalog failures — find what SUCCEEDED after
-each failure. The diff between failed input and successful input is the
-actual learning.
+Pipeline: Scanner (events) → Digest Builder → LLM → Recommendations
 
-All analyzers work on normalized ToolCall sequences. They are tool-agnostic:
-same analysis works for Claude Code, Cursor, Codex, or any agent with tool calls.
+No regex patterns, no static lookback windows, no hardcoded heuristics.
+A single LLM call understands the full conversation context and produces
+structured recommendations for CLAUDE.md / MEMORY.md.
+
+Supports any LLM provider via LiteLLM: Anthropic, OpenAI, Google, Bedrock,
+Ollama, and 100+ others. Auto-detects the best available model from env vars.
 """
 
 from __future__ import annotations
 
+import json
+import logging
 import os
-import re
-from collections import Counter, defaultdict
 
 from .models import (
-    AnalysisReport,
-    CommandPattern,
-    Correction,
-    EnvironmentFact,
-    ErrorCategory,
+    AnalysisResult,
     ProjectInfo,
-    RetryPattern,
+    Recommendation,
+    RecommendationTarget,
     SessionData,
-    StructureNote,
+    SessionEvent,
     ToolCall,
 )
 
-# How many messages ahead to look for a success after a failure
-_CORRECTION_WINDOW = 10
+logger = logging.getLogger(__name__)
 
+# Default models by provider (checked in order)
+_MODEL_DEFAULTS: list[tuple[str, str]] = [
+    ("ANTHROPIC_API_KEY", "claude-sonnet-4-6"),
+    ("OPENAI_API_KEY", "gpt-4o"),
+    ("GEMINI_API_KEY", "gemini/gemini-2.0-flash"),
+]
+
+_MAX_DIGEST_TOKENS = 80_000  # Budget for the digest (leave room for prompt + output)
+
+
+def _detect_default_model() -> str:
+    """Pick the best available model based on which API keys are set."""
+    for env_var, model in _MODEL_DEFAULTS:
+        if os.environ.get(env_var):
+            return model
+    raise RuntimeError(
+        "No LLM API key found. headroom learn needs one of:\n"
+        "  export ANTHROPIC_API_KEY=sk-ant-...   → uses claude-sonnet-4-6\n"
+        "  export OPENAI_API_KEY=sk-...          → uses gpt-4o\n"
+        "  export GEMINI_API_KEY=...             → uses gemini-2.0-flash\n"
+        "Or specify a model directly: headroom learn --model <litellm-model-name>"
+    )
 
-class FailureAnalyzer:
-    """Runs all analyzers on tool call data and produces an AnalysisReport."""
 
-    def analyze(self, project: ProjectInfo, sessions: list[SessionData]) -> AnalysisReport:
+class SessionAnalyzer:
+    """Analyzes session data via LLM to produce actionable recommendations.
+
+    Uses LiteLLM for provider-agnostic access to 100+ models.
+    Auto-detects the best available model from environment API keys.
+    """
+
+    def __init__(self, model: str | None = None):
+        self.model = model
+
+    def analyze(
+        self, project: ProjectInfo, sessions: list[SessionData]
+    ) -> AnalysisResult:
+        """Analyze sessions and produce recommendations via LLM."""
         all_calls = [tc for s in sessions for tc in s.tool_calls]
         failed_calls = [tc for tc in all_calls if tc.is_error]
 
-        report = AnalysisReport(
+        result = AnalysisResult(
             project=project,
+            total_sessions=len(sessions),
             total_calls=len(all_calls),
             total_failures=len(failed_calls),
-            total_sessions=len(sessions),
-            waste_bytes=sum(tc.output_bytes for tc in failed_calls),
         )
 
-        # Phase 1: Extract failure→success corrections (the core learning)
-        report.corrections = _extract_corrections(sessions)
-
-        # Phase 2: Analyze specific dimensions using corrections + raw data
-        report.environment_facts = _analyze_environment(sessions)
-        report.structure_notes = _analyze_structure(sessions, report.corrections)
-        report.command_patterns = _analyze_commands(sessions, report.corrections)
-        report.retry_patterns = _analyze_retries(sessions, report.corrections)
-        report.permission_issues = _analyze_permissions(sessions)
-        report.cross_session_patterns = _analyze_cross_session(sessions)
-
-        return report
+        if not failed_calls and not any(s.events for s in sessions):
+            return result
 
+        # Build compact digest of all sessions
+        digest = _build_digest(project, sessions)
 
-# =============================================================================
-# Success Correlation: The Core Learning Primitive
-# =============================================================================
+        # Resolve model (auto-detect if not specified)
+        model = self.model or _detect_default_model()
 
+        # Call LLM for analysis
+        try:
+            raw = _call_llm(digest, model)
+            result.recommendations = _parse_llm_response(raw)
+        except Exception as e:
+            logger.warning("LLM analysis failed: %s", e)
+            # Return result with stats but no recommendations
 
-def _extract_corrections(sessions: list[SessionData]) -> list[Correction]:
-    """For each failure, find the next success of the same tool type.
-
-    The pair (failed_input, success_input) is a Correction — the model
-    learned something and corrected its approach.
-    """
-    corrections: list[Correction] = []
-
-    for session in sessions:
-        calls = session.tool_calls
-        for i, tc in enumerate(calls):
-            if not tc.is_error:
-                continue
-            # Skip sibling errors (cascades, not real failures)
-            if tc.error_category == ErrorCategory.SIBLING_ERROR:
-                continue
-            # Look forward for a success of the same tool
-            for j in range(i + 1, min(i + _CORRECTION_WINDOW, len(calls))):
-                candidate = calls[j]
-                if candidate.name != tc.name or candidate.is_error:
-                    continue
-                # Found a success — is the input meaningfully different?
-                if candidate.input_data == tc.input_data:
-                    continue  # Exact same input succeeded (transient error)
-                corrections.append(
-                    Correction(
-                        tool_name=tc.name,
-                        failed_input=tc.input_data,
-                        success_input=candidate.input_data,
-                        error_category=tc.error_category,
-                        session_id=session.session_id,
-                    )
-                )
-                break
-
-    return corrections
+        return result
 
 
 # =============================================================================
-# Environment Analyzer (uses corrections for python/build tool detection)
+# Digest Builder — compact text representation of session events
 # =============================================================================
 
 
-def _analyze_environment(sessions: list[SessionData]) -> list[EnvironmentFact]:
-    """Detect which runtime commands work vs fail."""
-    facts: list[EnvironmentFact] = []
+def _build_digest(project: ProjectInfo, sessions: list[SessionData]) -> str:
+    """Build a token-efficient text digest of all session events.
 
-    python_failures: Counter[str] = Counter()
-    python_successes: Counter[str] = Counter()
-    python_sessions: dict[str, set[str]] = defaultdict(set)
+    The digest includes:
+    - Project context
+    - Per-session summaries with condensed event streams
+    - Error outputs (truncated), success indicators, user messages
+    """
+    lines: list[str] = []
+
+    # Project header
+    lines.append(f"Project: {project.name} ({project.project_path})")
+    total_calls = sum(len(s.tool_calls) for s in sessions)
+    total_failures = sum(s.failure_count for s in sessions)
+    total_tokens_in = sum(s.total_input_tokens for s in sessions)
+    total_tokens_out = sum(s.total_output_tokens for s in sessions)
+    lines.append(
+        f"Total: {len(sessions)} sessions, {total_calls} tool calls, "
+        f"{total_failures} failures ({total_failures / total_calls:.1%})"
+        if total_calls
+        else f"Total: {len(sessions)} sessions, 0 tool calls"
+    )
+    if total_tokens_in:
+        lines.append(f"Tokens used: {total_tokens_in:,} in / {total_tokens_out:,} out")
+    lines.append("")
+
+    # Budget tracking — stop adding events when we approach the limit
+    # Rough estimate: 4 chars per token
+    char_budget = _MAX_DIGEST_TOKENS * 4
+    chars_used = sum(len(ln) for ln in lines)
 
     for session in sessions:
-        for tc in session.tool_calls:
-            if tc.name not in ("Bash", "bash"):
-                continue
-            cmd = tc.input_data.get("command", "")
-            if not cmd:
-                continue
-
-            if tc.is_error and tc.error_category == ErrorCategory.MODULE_NOT_FOUND:
-                prefix = _extract_python_command(cmd)
-                if prefix:
-                    python_failures[prefix] += 1
-                    python_sessions[prefix].add(session.session_id)
-            elif not tc.is_error:
-                prefix = _extract_python_command(cmd)
-                if prefix:
-                    python_successes[prefix] += 1
-
-    if python_failures:
-        wrong = sorted(python_failures.keys(), key=lambda x: -python_failures[x])
-        correct = None
-        for cmd, _count in python_successes.most_common():
-            if cmd not in python_failures or python_successes[cmd] > python_failures[cmd] * 2:
-                correct = cmd
-                break
-        if correct and wrong:
-            total_evidence = sum(python_failures[w] for w in wrong)
-            total_sessions = len(set().union(*(python_sessions[w] for w in wrong)))
-            facts.append(
-                EnvironmentFact(
-                    category="python",
-                    correct_command=correct,
-                    wrong_commands=wrong[:5],
-                    evidence_count=total_evidence,
-                    sessions_seen=total_sessions,
-                )
-            )
+        if chars_used > char_budget:
+            lines.append(f"... (remaining {len(sessions) - sessions.index(session)} sessions truncated)")
+            break
 
-    return facts
+        session_header = (
+            f"=== Session {session.session_id[:12]} "
+            f"({len(session.tool_calls)} calls, {session.failure_count} failures"
+        )
+        if session.total_input_tokens:
+            session_header += f", {session.total_input_tokens:,} input tokens"
+        session_header += ") ==="
+        lines.append(session_header)
+        chars_used += len(session_header)
+
+        # Use events if available (richer context), fall back to tool_calls
+        if session.events:
+            for event in session.events:
+                if chars_used > char_budget:
+                    lines.append("  ... (remaining events truncated)")
+                    break
+                event_line = _format_event(event)
+                if event_line:
+                    lines.append(event_line)
+                    chars_used += len(event_line)
+        else:
+            for tc in session.tool_calls:
+                if chars_used > char_budget:
+                    lines.append("  ... (remaining calls truncated)")
+                    break
+                tc_line = _format_tool_call(tc)
+                lines.append(tc_line)
+                chars_used += len(tc_line)
 
+        lines.append("")
 
-# =============================================================================
-# Structure Analyzer (uses corrections to learn correct paths)
-# =============================================================================
+    return "\n".join(lines)
 
 
-def _analyze_structure(
-    sessions: list[SessionData], corrections: list[Correction]
-) -> list[StructureNote]:
-    """Find file structure issues and learn correct paths from corrections."""
-    notes: list[StructureNote] = []
+def _format_event(event: SessionEvent) -> str | None:
+    """Format a single event into a compact digest line."""
 
-    # 1. Path corrections: wrong path → correct path (from success correlation)
-    path_corrections: dict[str, Counter[str]] = defaultdict(Counter)
-    for c in corrections:
-        if c.error_category != ErrorCategory.FILE_NOT_FOUND:
-            continue
-        # Extract file paths from Read tool or from Bash commands (Codex uses shell for everything)
-        if c.tool_name in ("Read", "read"):
-            failed_path = c.failed_input.get("file_path", "")
-            success_path = c.success_input.get("file_path", "")
-        elif c.tool_name in ("Bash", "bash"):
-            failed_path = _extract_path_from_command(c.failed_input.get("command", ""))
-            success_path = _extract_path_from_command(c.success_input.get("command", ""))
-        else:
-            continue
-        if failed_path and success_path and failed_path != success_path:
-            path_corrections[failed_path][success_path] += 1
-
-    for wrong_path, correct_paths in path_corrections.items():
-        best_correct, count = correct_paths.most_common(1)[0]
-        # Make paths relative to project for readability
-        wrong_short = _shorten_path(wrong_path)
-        correct_short = _shorten_path(best_correct)
-        notes.append(
-            StructureNote(
-                category="path_correction",
-                path=wrong_short,
-                correct_path=correct_short,
-                note=f"Not at `{wrong_short}` → actually at `{correct_short}`",
-                evidence_count=count,
-            )
-        )
+    if event.type == "tool_call" and event.tool_call:
+        return _format_tool_call(event.tool_call)
 
-    # 2. Grep scope corrections: narrow path → broader path worked
-    scope_corrections: dict[str, Counter[str]] = defaultdict(Counter)
-    for c in corrections:
-        if c.tool_name not in ("Grep", "grep"):
-            continue
-        failed_path = c.failed_input.get("path", "")
-        success_path = c.success_input.get("path", "")
-        if failed_path and success_path and failed_path != success_path:
-            scope_corrections[_shorten_path(failed_path)][_shorten_path(success_path)] += 1
-
-    for wrong_scope, correct_scopes in scope_corrections.items():
-        best_scope, count = correct_scopes.most_common(1)[0]
-        notes.append(
-            StructureNote(
-                category="search_scope",
-                path=wrong_scope,
-                correct_path=best_scope,
-                note=f"Grep fails at `{wrong_scope}` → use `{best_scope}` instead",
-                evidence_count=count,
-            )
-        )
+    if event.type == "user_message" and event.text.strip():
+        text = event.text.strip()[:300]
+        return f"  [{event.msg_index}] USER: \"{text}\""
 
-    # 3. Large files (from raw failures, no correction needed)
-    large_files: Counter[str] = Counter()
-    large_sessions: dict[str, set[str]] = defaultdict(set)
-    for session in sessions:
-        for tc in session.tool_calls:
-            if (
-                tc.name in ("Read", "read")
-                and tc.is_error
-                and tc.error_category == ErrorCategory.FILE_TOO_LARGE
-            ):
-                path = tc.input_data.get("file_path", "")
-                if path:
-                    short = _shorten_path(path)
-                    large_files[short] += 1
-                    large_sessions[short].add(session.session_id)
-
-    for path, count in large_files.most_common(10):
-        if count < 2:
-            break
-        notes.append(
-            StructureNote(
-                category="large_file",
-                path=path,
-                note=f"Too large for full read — always use offset/limit ({count} failures, {len(large_sessions[path])} sessions)",
-                evidence_count=count,
-                sessions_seen=len(large_sessions[path]),
-            )
-        )
+    if event.type == "interruption":
+        return f"  [{event.msg_index}] INTERRUPTED: {event.text[:150]}"
 
-    # 4. Persistent missing paths (no correction found — file truly doesn't exist)
-    missing_no_correction: Counter[str] = Counter()
-    missing_sessions: dict[str, set[str]] = defaultdict(set)
-    corrected_paths = set(path_corrections.keys())
-    for session in sessions:
-        for tc in session.tool_calls:
-            if (
-                tc.name in ("Read", "read")
-                and tc.is_error
-                and tc.error_category == ErrorCategory.FILE_NOT_FOUND
-            ):
-                path = tc.input_data.get("file_path", "")
-                if path and path not in corrected_paths:
-                    missing_no_correction[path] += 1
-                    missing_sessions[path].add(session.session_id)
-
-    for path, count in missing_no_correction.most_common(10):
-        if count < 2:
-            break
-        short = _shorten_path(path)
-        notes.append(
-            StructureNote(
-                category="missing_path",
-                path=short,
-                note=f"Does not exist ({count} attempts, {len(missing_sessions[path])} sessions)",
-                evidence_count=count,
-                sessions_seen=len(missing_sessions[path]),
-            )
+    if event.type == "agent_summary":
+        return (
+            f"  [{event.msg_index}] SUBAGENT: {event.agent_tool_count} tool calls, "
+            f"{event.agent_tokens:,} tokens, {event.agent_duration_ms / 1000:.1f}s "
+            f"— prompt: \"{event.agent_prompt[:100]}\""
         )
 
-    return notes
+    return None
 
 
-# =============================================================================
-# Command Pattern Analyzer (uses corrections to learn command patterns)
-# =============================================================================
-
-
-def _analyze_commands(
-    sessions: list[SessionData], corrections: list[Correction]
-) -> list[CommandPattern]:
-    """Learn specific command patterns from Bash failure→success corrections."""
-    patterns: list[CommandPattern] = []
-
-    # Analyze Bash corrections
-    bash_corrections = [c for c in corrections if c.tool_name in ("Bash", "bash")]
-
-    # Group by error category to find patterns
-    by_category: dict[ErrorCategory, list[Correction]] = defaultdict(list)
-    for c in bash_corrections:
-        by_category[c.error_category].append(c)
-
-    # User-rejected commands: model should suggest, not execute
-    rejected = by_category.get(ErrorCategory.USER_REJECTED, [])
-    if rejected:
-        # Find the most commonly rejected command patterns
-        rejected_cmds: Counter[str] = Counter()
-        for c in rejected:
-            cmd = c.failed_input.get("command", "")
-            base = _extract_command_signature(cmd)
-            if base:
-                rejected_cmds[base] += 1
-
-        for cmd_sig, count in rejected_cmds.most_common(5):
-            if count < 2:
-                break
-            patterns.append(
-                CommandPattern(
-                    category="user_prefers_manual",
-                    wrong_pattern=f"Executing: {cmd_sig}",
-                    correct_pattern="Show the command to the user and let them run it",
-                    explanation=f"User rejected this command {count} times — they prefer to run it themselves",
-                    evidence_count=count,
-                    sessions_seen=len(
-                        {
-                            c.session_id
-                            for c in rejected
-                            if _extract_command_signature(c.failed_input.get("command", ""))
-                            == cmd_sig
-                        }
-                    ),
-                )
-            )
+def _format_tool_call(tc: ToolCall) -> str:
+    """Format a single tool call into a compact digest line."""
+    status = "ERROR" if tc.is_error else "OK"
+    error_cat = f"({tc.error_category.value})" if tc.is_error else ""
 
-    # Build failures: learn what command form works
-    build_fails = by_category.get(ErrorCategory.BUILD_FAILURE, [])
-    for c in build_fails:
-        failed_cmd = c.failed_input.get("command", "")
-        success_cmd = c.success_input.get("command", "")
-        if failed_cmd and success_cmd:
-            patterns.append(
-                CommandPattern(
-                    category="build",
-                    wrong_pattern=_extract_command_signature(failed_cmd),
-                    correct_pattern=_extract_command_signature(success_cmd),
-                    explanation="Build failed with first form, succeeded with second",
-                    evidence_count=1,
-                )
-            )
+    # Input summary
+    input_str = tc.input_summary[:120]
 
-    # Module not found: learn correct python invocation
-    module_fails = by_category.get(ErrorCategory.MODULE_NOT_FOUND, [])
-    if module_fails:
-        wrong_pythons: Counter[str] = Counter()
-        correct_pythons: Counter[str] = Counter()
-        for c in module_fails:
-            wp = _extract_python_command(c.failed_input.get("command", ""))
-            cp = _extract_python_command(c.success_input.get("command", ""))
-            if wp:
-                wrong_pythons[wp] += 1
-            if cp:
-                correct_pythons[cp] += 1
-
-        if wrong_pythons and correct_pythons:
-            wrong = wrong_pythons.most_common(1)[0][0]
-            correct = correct_pythons.most_common(1)[0][0]
-            if wrong != correct:
-                patterns.append(
-                    CommandPattern(
-                        category="python_runtime",
-                        wrong_pattern=f"`{wrong}` (modules not available)",
-                        correct_pattern=f"`{correct}` (has project dependencies)",
-                        explanation=f"Using `{wrong}` causes ModuleNotFoundError — use `{correct}` which has the project's venv",
-                        evidence_count=sum(wrong_pythons.values()),
-                    )
-                )
-
-    # Deduplicate patterns
-    seen = set()
-    unique = []
-    for p in patterns:
-        key = (p.category, p.wrong_pattern[:50])
-        if key not in seen:
-            seen.add(key)
-            unique.append(p)
-    return unique
+    if tc.is_error:
+        # Include truncated error output for failures
+        output_preview = tc.output[:200].replace("\n", " ").strip()
+        return f"  [{tc.msg_index}] {tc.name}: {input_str} → {status}{error_cat}: {output_preview}"
+    else:
+        # Just indicate success with size
+        size = f"({tc.output_bytes} bytes)" if tc.output_bytes > 0 else ""
+        return f"  [{tc.msg_index}] {tc.name}: {input_str} → {status} {size}"
 
 
 # =============================================================================
-# Retry Analyzer (uses corrections to provide specific suggestions)
+# LLM Call — Sonnet 4.6 with structured output
 # =============================================================================
 
+_SYSTEM_PROMPT = """\
+You are an expert at analyzing coding agent sessions to extract actionable patterns.
+
+You will receive a digest of tool call sessions from a coding agent (Claude Code, Codex, etc.).
+Your job is to identify patterns that, if documented, would PREVENT TOKEN WASTE in future sessions.
+
+Focus on:
+1. **Environment rules** — what runtime commands work vs fail (e.g., "use uv run python, not python3")
+2. **File structure facts** — known large files, correct paths, search scopes
+3. **User preferences** — things the user corrected, rejected, or explicitly requested
+4. **Failure patterns** — repeated failures that could be prevented with upfront knowledge
+5. **Workflow rules** — subagent guidance, command execution preferences
+6. **Token waste hotspots** — patterns that waste the most tokens (re-reads, wrong paths, retries)
+
+Rules:
+- Only include patterns with CLEAR evidence from the data (2+ occurrences or explicit user direction)
+- Every recommendation must be specific and actionable (not "be careful" but "use X instead of Y")
+- Estimate tokens saved per recommendation (how many tokens would be saved per session if this rule existed)
+- Separate stable project facts (CONTEXT_FILE) from evolving preferences (MEMORY_FILE)
+- CONTEXT_FILE rules go in CLAUDE.md/AGENTS.md — they are project-level, stable facts
+- MEMORY_FILE rules go in MEMORY.md — they are session-level, evolving preferences
+- Keep recommendations concise — each should be 1-3 lines of markdown
+- Do NOT produce tautological rules (e.g., "use python3 not python3")
+- Do NOT produce rules about things that only happened once (transient errors)
+
+Return ONLY valid JSON matching this schema — no other text:
+{
+  "context_file_rules": [
+    {
+      "section": "string — section heading (e.g., 'Environment', 'File Paths', 'Commands')",
+      "content": "string — markdown content, 1-3 bullet points",
+      "estimated_tokens_saved": "integer — tokens saved per session if rule existed",
+      "evidence_count": "integer — number of occurrences supporting this rule"
+    }
+  ],
+  "memory_file_rules": [
+    {
+      "section": "string — section heading",
+      "content": "string — markdown content, 1-3 bullet points",
+      "estimated_tokens_saved": "integer",
+      "evidence_count": "integer"
+    }
+  ]
+}
+"""
 
-def _analyze_retries(
-    sessions: list[SessionData], corrections: list[Correction]
-) -> list[RetryPattern]:
-    """Find stubborn retries with specific fix suggestions from corrections."""
-    patterns: list[RetryPattern] = []
-
-    # Build a correction lookup: (tool, error_category) → list of corrections
-    correction_lookup: dict[tuple[str, str], list[Correction]] = defaultdict(list)
-    for c in corrections:
-        correction_lookup[(c.tool_name, c.error_category.value)].append(c)
-
-    # Find retry streaks
-    pattern_counter: Counter[tuple[str, str, str]] = Counter()
-    max_retries: dict[tuple[str, str, str], int] = {}
 
-    for session in sessions:
-        streak: dict[str, list[ToolCall]] = defaultdict(list)
-        for tc in session.tool_calls:
-            key = f"{tc.name}:{tc.error_category.value}"
-            if tc.is_error:
-                streak[key].append(tc)
-            else:
-                if len(streak.get(key, [])) >= 3:
-                    calls = streak[key]
-                    pk = (tc.name, calls[0].error_category.value, calls[0].input_summary[:50])
-                    pattern_counter[pk] += 1
-                    max_retries[pk] = max(max_retries.get(pk, 0), len(calls))
-                streak[key] = []
-        for _key, calls in streak.items():
-            if len(calls) >= 3:
-                pk = (calls[0].name, calls[0].error_category.value, calls[0].input_summary[:50])
-                pattern_counter[pk] += 1
-                max_retries[pk] = max(max_retries.get(pk, 0), len(calls))
-
-    for (tool, err_cat, input_key), count in pattern_counter.most_common(10):
-        max_r = max_retries.get((tool, err_cat, input_key), 3)
-
-        # Try to get a SPECIFIC suggestion from corrections
-        relevant_corrections = correction_lookup.get((tool, err_cat), [])
-        suggestion = _build_specific_suggestion(tool, err_cat, relevant_corrections)
-
-        patterns.append(
-            RetryPattern(
-                tool_name=tool,
-                error_category=ErrorCategory(err_cat),
-                description=f"{tool} failing with {err_cat}: {input_key}",
-                max_retries_seen=max_r,
-                suggestion=suggestion,
-                evidence_count=count,
-            )
-        )
+def _call_llm(digest: str, model: str) -> dict:
+    """Call LLM with the session digest and return parsed JSON.
 
-    return patterns
-
-
-def _build_specific_suggestion(
-    tool: str, error_category: str, corrections: list[Correction]
-) -> str:
-    """Build a specific suggestion from actual corrections, not generic advice."""
-    if not corrections:
-        # No corrections available — use tool+error specific defaults
-        return _default_suggestion(tool, error_category)
-
-    # Summarize what corrections tell us
-    if tool in ("Read", "read") and error_category == "file_not_found":
-        examples = []
-        for c in corrections[:3]:
-            wrong = _shorten_path(c.failed_input.get("file_path", ""))
-            right = _shorten_path(c.success_input.get("file_path", ""))
-            if wrong and right:
-                examples.append(f"`{wrong}` → `{right}`")
-        if examples:
-            return "Use Glob to discover actual path. Known corrections: " + "; ".join(examples)
-
-    if tool in ("Grep", "grep"):
-        # Summarize scope corrections
-        scopes = set()
-        for c in corrections[:5]:
-            right_path = c.success_input.get("path", "")
-            if right_path:
-                scopes.add(_shorten_path(right_path))
-        if scopes:
-            return f"Scope searches to: {', '.join(sorted(scopes)[:3])}"
-
-    if tool in ("Bash", "bash") and error_category == "user_rejected":
-        return "User prefers to run this command themselves. Show the command, don't execute it."
-
-    if tool in ("Bash", "bash") and error_category == "module_not_found":
-        correct_cmds = set()
-        for c in corrections[:5]:
-            prefix = _extract_python_command(c.success_input.get("command", ""))
-            if prefix:
-                correct_cmds.add(prefix)
-        if correct_cmds:
-            return f"Use {' or '.join(sorted(correct_cmds))} (has project dependencies)"
-
-    # Fallback: show one correction example
-    c = corrections[0]
-    return f"What worked: {c.success_summary[:80]}"
-
-
-def _default_suggestion(tool: str, error_category: str) -> str:
-    """Fallback when no corrections are available."""
-    defaults = {
-        (
-            "Glob",
-            "no_matches",
-        ): "Broaden pattern to **/*.ext or use ls to explore directory structure",
-        ("Grep", "no_matches"): "Try case-insensitive (-i) or broaden search scope",
-        ("Grep", "timeout"): "Scope Grep to a specific subdirectory — the full repo is too large",
-        ("Read", "file_not_found"): "Use Glob to discover the file path before Read",
-        ("Read", "file_too_large"): "Use offset/limit parameters for this file",
-        ("Bash", "module_not_found"): "Use the project's virtualenv Python",
-        ("Bash", "permission_denied"): "Do not retry — try a different approach",
-        ("Bash", "command_not_found"): "Verify tool is installed: which <tool>",
-        ("Bash", "user_rejected"): "User does not want this command executed. Show it instead.",
-        ("Edit", "unknown"): "If old_string has multiple matches, add more surrounding context",
-    }
-    return defaults.get((tool, error_category), "Try an alternative approach after 2 failures")
+    Uses LiteLLM for provider-agnostic access. The model string determines
+    the provider: "claude-*" → Anthropic, "gpt-*" → OpenAI, "gemini/*" → Google, etc.
+    """
+    import litellm
+
+    # Suppress LiteLLM's verbose logging
+    litellm.suppress_debug_info = True
+
+    # For Anthropic models, bypass ANTHROPIC_BASE_URL which may point to
+    # the user's local headroom proxy
+    api_base = None
+    if model.startswith("claude"):
+        api_base = "https://api.anthropic.com"
+
+    response = litellm.completion(
+        model=model,
+        messages=[
+            {"role": "system", "content": _SYSTEM_PROMPT},
+            {
+                "role": "user",
+                "content": (
+                    "Analyze these coding agent sessions and return JSON recommendations:\n\n"
+                    + digest
+                ),
+            },
+        ],
+        max_tokens=4096,
+        api_base=api_base,
+    )
+
+    # Extract text from response
+    text = response.choices[0].message.content or ""
+
+    # Parse JSON — handle both raw JSON and ```json fenced blocks
+    text = text.strip()
+    if text.startswith("```"):
+        lines = text.split("\n")
+        lines = [ln for ln in lines[1:] if not ln.strip().startswith("```")]
+        text = "\n".join(lines)
+
+    result: dict = json.loads(text)
+    return result
 
 
 # =============================================================================
-# Permission Analyzer
+# Response Parser — LLM JSON → Recommendation list
 # =============================================================================
 
 
-def _analyze_permissions(sessions: list[SessionData]) -> list[str]:
-    """Find commands repeatedly denied — with specific advice."""
-    denied: Counter[str] = Counter()
-    denied_cmds: dict[str, str] = {}  # key → full command example
+def _parse_llm_response(raw: dict) -> list[Recommendation]:
+    """Convert LLM structured output into Recommendation objects."""
+    recommendations: list[Recommendation] = []
 
-    for session in sessions:
-        for tc in session.tool_calls:
-            if not tc.is_error:
-                continue
-            if tc.error_category not in (
-                ErrorCategory.PERMISSION_DENIED,
-                ErrorCategory.USER_REJECTED,
-            ):
-                continue
-            sig = _extract_command_signature(
-                tc.input_data.get("command", "")
-                if tc.name in ("Bash", "bash")
-                else tc.input_summary
+    for rule in raw.get("context_file_rules", []):
+        if not isinstance(rule, dict):
+            continue
+        section = rule.get("section", "").strip()
+        content = rule.get("content", "").strip()
+        if not section or not content:
+            continue
+        recommendations.append(
+            Recommendation(
+                target=RecommendationTarget.CONTEXT_FILE,
+                section=section,
+                content=content,
+                confidence=0.9,
+                evidence_count=_safe_int(rule.get("evidence_count", 1)),
+                estimated_tokens_saved=_safe_int(rule.get("estimated_tokens_saved", 0)),
             )
-            key = f"{tc.name}: {sig}"
-            denied[key] += 1
-            if key not in denied_cmds:
-                denied_cmds[key] = tc.input_summary[:80]
-
-    results = []
-    for key, count in denied.most_common(10):
-        if count < 3:
-            break
-        results.append(
-            f"{key} — denied {count} times. Show the command to the user instead of executing it."
         )
-    return results
-
 
-# =============================================================================
-# Cross-Session Analyzer
-# =============================================================================
+    for rule in raw.get("memory_file_rules", []):
+        if not isinstance(rule, dict):
+            continue
+        section = rule.get("section", "").strip()
+        content = rule.get("content", "").strip()
+        if not section or not content:
+            continue
+        recommendations.append(
+            Recommendation(
+                target=RecommendationTarget.MEMORY_FILE,
+                section=section,
+                content=content,
+                confidence=0.7,
+                evidence_count=_safe_int(rule.get("evidence_count", 1)),
+                estimated_tokens_saved=_safe_int(rule.get("estimated_tokens_saved", 0)),
+            )
+        )
 
+    # Sort by estimated token savings
+    recommendations.sort(key=lambda r: r.estimated_tokens_saved, reverse=True)
 
-def _analyze_cross_session(sessions: list[SessionData]) -> list[str]:
-    """Find failure patterns that repeat across 3+ sessions."""
-    pattern_sessions: dict[str, set[str]] = defaultdict(set)
+    return recommendations
 
-    for session in sessions:
-        for tc in session.tool_calls:
-            if not tc.is_error or tc.error_category == ErrorCategory.SIBLING_ERROR:
-                continue
-            key = f"{tc.name}|{tc.error_category.value}|{tc.input_summary[:60]}"
-            pattern_sessions[key].add(session.session_id)
-
-    cross_session = []
-    for key, session_ids in sorted(pattern_sessions.items(), key=lambda x: -len(x[1])):
-        if len(session_ids) < 3:
-            continue
-        parts = key.split("|", 2)
-        tool, err, inp = parts[0], parts[1], parts[2] if len(parts) > 2 else "?"
-        cross_session.append(f"{tool} {err}: {inp} (across {len(session_ids)} sessions)")
-        if len(cross_session) >= 15:
-            break
 
-    return cross_session
+def _safe_int(val: object) -> int:
+    """Safely convert a value to int."""
+    if isinstance(val, int):
+        return val
+    if isinstance(val, (float, str)):
+        try:
+            return int(val)
+        except (ValueError, TypeError):
+            return 0
+    return 0
 
 
 # =============================================================================
-# Helpers
+# Legacy compatibility alias
 # =============================================================================
 
 
-_PYTHON_CMD_RE = re.compile(
-    r"^((?:source\s+\S+\s*&&\s*)?(?:\.venv/bin/)?(?:python3?|uv run python|uv run|/opt/nflx/python))"
-    r"(?:\s|$|-)"  # Must be followed by space, end, or dash (python3 -c, python -)
-)
+class FailureAnalyzer:
+    """Legacy alias for SessionAnalyzer — used by existing CLI code."""
 
+    def __init__(self) -> None:
+        self._analyzer = SessionAnalyzer()
 
-def _extract_python_command(cmd: str) -> str | None:
-    """Extract the python invocation prefix from a command."""
-    cmd = cmd.strip()
-    if "&&" in cmd:
-        for part in cmd.split("&&"):
-            result = _extract_python_command(part.strip())
-            if result:
-                return result
-        return None
-    m = _PYTHON_CMD_RE.match(cmd)
-    return m.group(1) if m else None
-
-
-def _extract_command_signature(cmd: str) -> str:
-    """Extract a normalizable command signature (first ~60 chars, no args)."""
-    cmd = cmd.strip()
-    # Truncate at first newline
-    if "\n" in cmd:
-        cmd = cmd.split("\n")[0]
-    return cmd[:60]
-
-
-def _extract_path_from_command(cmd: str) -> str:
-    """Extract a file path from a shell command (cat, sed, head, etc.)."""
-    cmd = cmd.strip()
-    # Look for path-like tokens (containing / and ending in a file extension)
-    path_re = re.compile(r"""(?:^|[\s'"])(/[^\s'"]+\.\w+)""")
-    match = path_re.search(cmd)
-    if match:
-        return match.group(1)
-    # Also try: last token that looks like a path
-    tokens = cmd.split()
-    for token in reversed(tokens):
-        if "/" in token and not token.startswith("-"):
-            return token.strip("'\"")
-    return ""
-
-
-def _shorten_path(path: str) -> str:
-    """Make a path relative to home for readability."""
-    home = os.path.expanduser("~")
-    if path.startswith(home):
-        return "~" + path[len(home) :]
-    return path
+    def analyze(
+        self, project: ProjectInfo, sessions: list[SessionData]
+    ) -> AnalysisResult:
+        return self._analyzer.analyze(project, sessions)

headroom/learn/models.py

@@ -76,13 +76,42 @@ class ToolCall:
         return str(self.input_data)[:80]
 
 
+@dataclass
+class SessionEvent:
+    """Any event in a session — tool calls, user messages, interruptions.
+
+    Provides richer context than ToolCall alone, enabling
+    user preference mining and conversation understanding.
+    """
+
+    type: str  # "tool_call", "user_message", "interruption", "agent_summary"
+    msg_index: int
+    timestamp: str | None = None
+
+    # For tool_call type
+    tool_call: ToolCall | None = None
+
+    # For user_message type
+    text: str = ""
+
+    # For agent_summary type (subagent results)
+    agent_id: str = ""
+    agent_tool_count: int = 0
+    agent_tokens: int = 0
+    agent_duration_ms: int = 0
+    agent_prompt: str = ""
+
+
 @dataclass
 class SessionData:
     """Normalized data from a single conversation session."""
 
     session_id: str
     tool_calls: list[ToolCall] = field(default_factory=list)
+    events: list[SessionEvent] = field(default_factory=list)
     timestamp: datetime | None = None
+    total_input_tokens: int = 0
+    total_output_tokens: int = 0
 
     @property
     def failure_count(self) -> int:
@@ -118,92 +147,6 @@ class RecommendationTarget(str, Enum):
     MEMORY_FILE = "memory_file"  # MEMORY.md or equivalent
 
 
-@dataclass
-class EnvironmentFact:
-    """A learned fact about the project's runtime environment."""
-
-    category: str  # "python", "build_tool", "test_runner", "linter"
-    correct_command: str  # What works: "uv run python"
-    wrong_commands: list[str] = field(default_factory=list)  # What fails: ["python3"]
-    evidence_count: int = 0  # How many failures support this
-    sessions_seen: int = 0  # Across how many sessions
-
-
-@dataclass
-class StructureNote:
-    """A learned fact about the project's file structure."""
-
-    category: str  # "large_file", "missing_path", "path_correction", "search_scope"
-    path: str  # The file path in question
-    note: str  # Human-readable note
-    correct_path: str = ""  # If corrected, what the actual path is
-    evidence_count: int = 0
-    sessions_seen: int = 0
-
-
-@dataclass
-class Correction:
-    """A failure→success pair: what failed and what worked instead.
-
-    This is the core learning primitive. By comparing the failed input to
-    the successful input, we extract specific actionable knowledge.
-    """
-
-    tool_name: str
-    failed_input: dict  # The input that failed
-    success_input: dict  # The input that succeeded
-    error_category: ErrorCategory
-    session_id: str
-
-    @property
-    def failed_summary(self) -> str:
-        if self.tool_name in ("Read", "read"):
-            return str(self.failed_input.get("file_path", "?"))
-        if self.tool_name in ("Bash", "bash"):
-            return str(self.failed_input.get("command", "?"))[:100]
-        if self.tool_name in ("Grep", "grep"):
-            path = str(self.failed_input.get("path", ""))
-            pattern = str(self.failed_input.get("pattern", ""))
-            return f"pattern={pattern[:40]} path={path}"
-        return str(self.failed_input)[:80]
-
-    @property
-    def success_summary(self) -> str:
-        if self.tool_name in ("Read", "read"):
-            return str(self.success_input.get("file_path", "?"))
-        if self.tool_name in ("Bash", "bash"):
-            return str(self.success_input.get("command", "?"))[:100]
-        if self.tool_name in ("Grep", "grep"):
-            path = str(self.success_input.get("path", ""))
-            pattern = str(self.success_input.get("pattern", ""))
-            return f"pattern={pattern[:40]} path={path}"
-        return str(self.success_input)[:80]
-
-
-@dataclass
-class CommandPattern:
-    """A learned pattern about how commands should be run in this project."""
-
-    category: str  # "gradle", "python", "test", "build", "lint"
-    wrong_pattern: str  # What fails (e.g., "cd /path && ./gradlew")
-    correct_pattern: str  # What works (e.g., "../gradlew from axion/")
-    explanation: str  # Why (e.g., "user rejects cd-based gradle, use relative path")
-    evidence_count: int = 0
-    sessions_seen: int = 0
-
-
-@dataclass
-class RetryPattern:
-    """A pattern of stubborn retries that should be prevented."""
-
-    tool_name: str
-    error_category: ErrorCategory
-    description: str  # What keeps failing
-    max_retries_seen: int  # Worst case observed
-    suggestion: str  # What to do instead (SPECIFIC, from success correlation)
-    evidence_count: int = 0
-
-
 @dataclass
 class Recommendation:
     """A concrete recommendation to write to a context/memory file."""
@@ -213,25 +156,18 @@ class Recommendation:
     content: str  # Markdown content for the section
     confidence: float = 0.0  # 0-1, based on evidence strength
     evidence_count: int = 0  # Number of failures supporting this
+    estimated_tokens_saved: int = 0  # Projected savings if recommendation is followed
 
 
 @dataclass
-class AnalysisReport:
-    """Complete output of failure analysis for a project."""
+class AnalysisResult:
+    """Output of session analysis — stats + recommendations."""
 
     project: ProjectInfo
+    total_sessions: int = 0
     total_calls: int = 0
     total_failures: int = 0
-    total_sessions: int = 0
-    waste_bytes: int = 0
-
-    environment_facts: list[EnvironmentFact] = field(default_factory=list)
-    structure_notes: list[StructureNote] = field(default_factory=list)
-    retry_patterns: list[RetryPattern] = field(default_factory=list)
-    command_patterns: list[CommandPattern] = field(default_factory=list)
-    corrections: list[Correction] = field(default_factory=list)
-    permission_issues: list[str] = field(default_factory=list)
-    cross_session_patterns: list[str] = field(default_factory=list)
+    recommendations: list[Recommendation] = field(default_factory=list)
 
     @property
     def failure_rate(self) -> float:

headroom/learn/scanner.py

@@ -16,6 +16,7 @@ from .models import (
     ErrorCategory,
     ProjectInfo,
     SessionData,
+    SessionEvent,
     ToolCall,
 )
 
@@ -213,6 +214,9 @@ class ClaudeCodeScanner(ConversationScanner):
         session_id = jsonl_path.stem
         tool_uses: dict[str, tuple[str, dict]] = {}  # tc_id → (tool_name, input)
         tool_calls: list[ToolCall] = []
+        events: list[SessionEvent] = []
+        total_input_tokens = 0
+        total_output_tokens = 0
         msg_index = 0
 
         try:
@@ -225,17 +229,43 @@ class ClaudeCodeScanner(ConversationScanner):
 
                     msg_index += 1
                     line_type = d.get("type", "")
+                    ts = d.get("timestamp", None)
 
                     if line_type == "assistant":
                         self._extract_tool_uses(d, tool_uses)
+                        # Extract token usage
+                        usage = d.get("message", {}).get("usage", {})
+                        total_input_tokens += usage.get("input_tokens", 0)
+                        total_input_tokens += usage.get("cache_read_input_tokens", 0)
+                        total_input_tokens += usage.get("cache_creation_input_tokens", 0)
+                        total_output_tokens += usage.get("output_tokens", 0)
                     elif line_type == "user":
-                        self._extract_tool_results(d, tool_uses, tool_calls, msg_index)
+                        self._extract_tool_results(
+                            d, tool_uses, tool_calls, events, msg_index, ts
+                        )
+                        self._extract_user_events(d, events, msg_index, ts)
 
         except (OSError, UnicodeDecodeError) as e:
             logger.debug("Failed to read %s: %s", jsonl_path, e)
             return None
 
-        return SessionData(session_id=session_id, tool_calls=tool_calls)
+        # Also wrap tool_calls as events for unified access
+        for tc in tool_calls:
+            if not any(
+                e.type == "tool_call" and e.tool_call is tc for e in events
+            ):
+                events.append(
+                    SessionEvent(type="tool_call", msg_index=tc.msg_index, tool_call=tc)
+                )
+        events.sort(key=lambda e: e.msg_index)
+
+        return SessionData(
+            session_id=session_id,
+            tool_calls=tool_calls,
+            events=events,
+            total_input_tokens=total_input_tokens,
+            total_output_tokens=total_output_tokens,
+        )
 
     def _extract_tool_uses(self, d: dict, tool_uses: dict[str, tuple[str, dict]]) -> None:
         """Extract tool_use blocks from an assistant message."""
@@ -258,7 +288,9 @@ class ClaudeCodeScanner(ConversationScanner):
         d: dict,
         tool_uses: dict[str, tuple[str, dict]],
         tool_calls: list[ToolCall],
+        events: list[SessionEvent],
         msg_index: int,
+        timestamp: str | None = None,
     ) -> None:
         """Extract tool_result blocks from a user message and match to tool_uses."""
         msg = d.get("message", {})
@@ -288,18 +320,77 @@ class ClaudeCodeScanner(ConversationScanner):
 
             error_cat = classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
 
-            tool_calls.append(
-                ToolCall(
-                    name=name,
-                    tool_call_id=tc_id,
-                    input_data=inp,
-                    output=result_content,
-                    is_error=is_err,
-                    error_category=error_cat,
+            tc = ToolCall(
+                name=name,
+                tool_call_id=tc_id,
+                input_data=inp,
+                output=result_content,
+                is_error=is_err,
+                error_category=error_cat,
+                msg_index=msg_index,
+                output_bytes=len(result_content.encode("utf-8")),
+            )
+            tool_calls.append(tc)
+            events.append(
+                SessionEvent(type="tool_call", msg_index=msg_index, timestamp=timestamp, tool_call=tc)
+            )
+
+            # Extract subagent summary from toolUseResult metadata
+            if name in ("Agent", "agent"):
+                tool_result_meta = d.get("toolUseResult", {})
+                if isinstance(tool_result_meta, dict):
+                    events.append(
+                        SessionEvent(
+                            type="agent_summary",
+                            msg_index=msg_index,
+                            timestamp=timestamp,
+                            agent_id=tool_result_meta.get("agentId", ""),
+                            agent_tool_count=tool_result_meta.get("totalToolUseCount", 0),
+                            agent_tokens=tool_result_meta.get("totalTokens", 0),
+                            agent_duration_ms=tool_result_meta.get("totalDurationMs", 0),
+                            agent_prompt=tool_result_meta.get("prompt", "")[:200],
+                        )
+                    )
+
+    def _extract_user_events(
+        self,
+        d: dict,
+        events: list[SessionEvent],
+        msg_index: int,
+        timestamp: str | None = None,
+    ) -> None:
+        """Extract user text messages and interruptions from a user line."""
+        msg = d.get("message", {})
+        content = msg.get("content", "")
+
+        # Human text messages have content as a string, not a list
+        if isinstance(content, str) and content.strip():
+            events.append(
+                SessionEvent(
+                    type="user_message",
                     msg_index=msg_index,
-                    output_bytes=len(result_content.encode("utf-8")),
+                    timestamp=timestamp,
+                    text=content[:500],
                 )
             )
+            return
+
+        # Check for interruptions in list-format content
+        if isinstance(content, list):
+            for block in content:
+                if not isinstance(block, dict):
+                    continue
+                if block.get("type") == "text":
+                    text = block.get("text", "")
+                    if "[Request interrupted by user" in text:
+                        events.append(
+                            SessionEvent(
+                                type="interruption",
+                                msg_index=msg_index,
+                                timestamp=timestamp,
+                                text=text[:200],
+                            )
+                        )
 
 
 def _decode_project_path(escaped_name: str) -> Path | None:

headroom/learn/writer.py

@@ -12,14 +12,9 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 from .models import (
-    AnalysisReport,
-    CommandPattern,
-    EnvironmentFact,
     ProjectInfo,
     Recommendation,
     RecommendationTarget,
-    RetryPattern,
-    StructureNote,
 )
 
 # Marker delimiters for Headroom-managed sections
@@ -31,213 +26,6 @@ _MARKER_PATTERN = re.compile(
 )
 
 
-# =============================================================================
-# Recommender: AnalysisReport → Recommendations
-# =============================================================================
-
-
-class Recommender:
-    """Converts an AnalysisReport into concrete markdown recommendations.
-
-    Quality gates:
-    - Each recommendation needs min_evidence failures to be included
-    - Total recommendations need min_total_evidence to trigger any file writes
-    - Low-confidence recommendations are filtered out
-    """
-
-    def __init__(
-        self,
-        min_evidence: int = 2,
-        min_confidence: float = 0.3,
-        min_total_evidence: int = 3,
-    ):
-        self.min_evidence = min_evidence
-        self.min_confidence = min_confidence
-        self.min_total_evidence = min_total_evidence
-
-    def recommend(self, report: AnalysisReport) -> list[Recommendation]:
-        recommendations: list[Recommendation] = []
-
-        # Environment facts → CONTEXT_FILE (CLAUDE.md)
-        if report.environment_facts:
-            content = self._format_environment(report.environment_facts)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.CONTEXT_FILE,
-                    section="Environment",
-                    content=content,
-                    confidence=min(
-                        1.0, sum(f.evidence_count for f in report.environment_facts) / 10
-                    ),
-                    evidence_count=sum(f.evidence_count for f in report.environment_facts),
-                )
-            )
-
-        # Large files → CONTEXT_FILE
-        large_files = [n for n in report.structure_notes if n.category == "large_file"]
-        if large_files:
-            content = self._format_large_files(large_files)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.CONTEXT_FILE,
-                    section="Known Large Files",
-                    content=content,
-                    confidence=min(1.0, sum(n.evidence_count for n in large_files) / 5),
-                    evidence_count=sum(n.evidence_count for n in large_files),
-                )
-            )
-
-        # Path corrections → CONTEXT_FILE (these are stable project structure facts)
-        path_corrections = [n for n in report.structure_notes if n.category == "path_correction"]
-        if path_corrections:
-            content = self._format_path_corrections(path_corrections)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.CONTEXT_FILE,
-                    section="File Path Corrections",
-                    content=content,
-                    confidence=0.9,
-                    evidence_count=sum(n.evidence_count for n in path_corrections),
-                )
-            )
-
-        # Search scope corrections → CONTEXT_FILE
-        scope_corrections = [n for n in report.structure_notes if n.category == "search_scope"]
-        if scope_corrections:
-            content = self._format_scope_corrections(scope_corrections)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.CONTEXT_FILE,
-                    section="Search Scope",
-                    content=content,
-                    confidence=0.8,
-                    evidence_count=sum(n.evidence_count for n in scope_corrections),
-                )
-            )
-
-        # Command patterns → CONTEXT_FILE (stable project-level facts)
-        if report.command_patterns:
-            content = self._format_command_patterns(report.command_patterns)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.CONTEXT_FILE,
-                    section="Command Patterns",
-                    content=content,
-                    confidence=0.9,
-                    evidence_count=sum(p.evidence_count for p in report.command_patterns),
-                )
-            )
-
-        # Missing paths (no correction found) → MEMORY_FILE
-        missing_paths = [n for n in report.structure_notes if n.category == "missing_path"]
-        if missing_paths:
-            content = self._format_missing_paths(missing_paths)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.MEMORY_FILE,
-                    section="Known Missing Paths",
-                    content=content,
-                    confidence=0.6,
-                    evidence_count=sum(n.evidence_count for n in missing_paths),
-                )
-            )
-
-        # Retry patterns (with specific suggestions) → MEMORY_FILE
-        if report.retry_patterns:
-            content = self._format_retry_patterns(report.retry_patterns)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.MEMORY_FILE,
-                    section="Retry Prevention",
-                    content=content,
-                    confidence=0.7,
-                    evidence_count=sum(p.evidence_count for p in report.retry_patterns),
-                )
-            )
-
-        # Permission issues → MEMORY_FILE
-        if report.permission_issues:
-            content = self._format_permissions(report.permission_issues)
-            recommendations.append(
-                Recommendation(
-                    target=RecommendationTarget.MEMORY_FILE,
-                    section="Permission Notes",
-                    content=content,
-                    confidence=0.5,
-                    evidence_count=len(report.permission_issues),
-                )
-            )
-
-        # Quality gate: filter out weak recommendations
-        recommendations = [
-            r
-            for r in recommendations
-            if r.evidence_count >= self.min_evidence and r.confidence >= self.min_confidence
-        ]
-
-        # Quality gate: if total evidence is too low, don't recommend anything
-        total_evidence = sum(r.evidence_count for r in recommendations)
-        if total_evidence < self.min_total_evidence:
-            return []
-
-        return recommendations
-
-    def _format_environment(self, facts: list[EnvironmentFact]) -> str:
-        lines = []
-        for fact in facts:
-            wrong = ", ".join(f"`{w}`" for w in fact.wrong_commands[:3])
-            lines.append(
-                f"- **{fact.category.title()}**: use `{fact.correct_command}` "
-                f"(not {wrong} — {fact.evidence_count} failures observed)"
-            )
-        return "\n".join(lines)
-
-    def _format_large_files(self, notes: list[StructureNote]) -> str:
-        lines = ["Always use `offset` and `limit` parameters with Read for these files:"]
-        for note in sorted(notes, key=lambda n: -n.evidence_count):
-            lines.append(f"- `{note.path}` ({note.note})")
-        return "\n".join(lines)
-
-    def _format_path_corrections(self, notes: list[StructureNote]) -> str:
-        lines = ["These file paths are commonly guessed wrong. Use the correct paths:"]
-        for note in sorted(notes, key=lambda n: -n.evidence_count):
-            lines.append(f"- `{note.path}` → actually at `{note.correct_path}`")
-        return "\n".join(lines)
-
-    def _format_scope_corrections(self, notes: list[StructureNote]) -> str:
-        lines = ["When searching, use these scopes (broader paths work, narrow ones fail):"]
-        for note in sorted(notes, key=lambda n: -n.evidence_count):
-            lines.append(f"- Don't search `{note.path}` → use `{note.correct_path}` instead")
-        return "\n".join(lines)
-
-    def _format_command_patterns(self, patterns: list[CommandPattern]) -> str:
-        lines = []
-        for p in sorted(patterns, key=lambda p: -p.evidence_count):
-            lines.append(f"- **{p.category}**: {p.explanation}")
-            lines.append(f"  - Wrong: {p.wrong_pattern}")
-            lines.append(f"  - Correct: {p.correct_pattern}")
-        return "\n".join(lines)
-
-    def _format_missing_paths(self, notes: list[StructureNote]) -> str:
-        lines = []
-        for note in sorted(notes, key=lambda n: -n.evidence_count):
-            lines.append(f"- `{note.path}` — {note.note}")
-        return "\n".join(lines)
-
-    def _format_retry_patterns(self, patterns: list[RetryPattern]) -> str:
-        lines = []
-        for p in sorted(patterns, key=lambda p: -p.evidence_count):
-            lines.append(f"- {p.description}")
-            lines.append(f"  → {p.suggestion}")
-        return "\n".join(lines)
-
-    def _format_permissions(self, issues: list[str]) -> str:
-        lines = []
-        for issue in issues:
-            lines.append(f"- {issue}")
-        return "\n".join(lines)
-
-
 # =============================================================================
 # Abstract Writer
 # =============================================================================
@@ -273,6 +61,42 @@ class WriteResult:
         self.content_by_file[path] = content
 
 
+# =============================================================================
+# Shared section builder
+# =============================================================================
+
+
+def _build_section(recommendations: list[Recommendation]) -> str:
+    """Build the marker-delimited section content from recommendations."""
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    lines = [
+        _MARKER_START,
+        "## Headroom Learned Patterns",
+        f"*Auto-generated by `headroom learn` on {now} — do not edit manually*",
+        "",
+    ]
+
+    for rec in recommendations:
+        lines.append(f"### {rec.section}")
+        if rec.estimated_tokens_saved > 0:
+            lines.append(f"*~{rec.estimated_tokens_saved:,} tokens/session saved*")
+        lines.append(rec.content)
+        lines.append("")
+
+    lines.append(_MARKER_END)
+    return "\n".join(lines)
+
+
+def _merge_into_file(file_path: Path, section: str) -> str:
+    """Merge the section into an existing file, replacing any prior section."""
+    if file_path.exists():
+        existing = file_path.read_text()
+        if _MARKER_START in existing:
+            return _MARKER_PATTERN.sub(section, existing)
+        return existing.rstrip() + "\n\n" + section + "\n"
+    return section + "\n"
+
+
 # =============================================================================
 # Claude Code Writer
 # =============================================================================
@@ -290,28 +114,23 @@ class ClaudeCodeWriter(ContextWriter):
         result = WriteResult()
         result.dry_run = dry_run
 
-        # Group recommendations by target
         context_recs = [r for r in recommendations if r.target == RecommendationTarget.CONTEXT_FILE]
         memory_recs = [r for r in recommendations if r.target == RecommendationTarget.MEMORY_FILE]
 
-        # Generate CLAUDE.md content
         if context_recs:
             claude_md_path = self._resolve_context_path(project)
-            section_content = self._build_section(context_recs)
-            full_content = self._merge_into_file(claude_md_path, section_content)
+            section_content = _build_section(context_recs)
+            full_content = _merge_into_file(claude_md_path, section_content)
             result.add(claude_md_path, full_content)
-
             if not dry_run:
                 claude_md_path.parent.mkdir(parents=True, exist_ok=True)
                 claude_md_path.write_text(full_content)
 
-        # Generate MEMORY.md content
         if memory_recs:
             memory_path = self._resolve_memory_path(project)
-            section_content = self._build_section(memory_recs)
-            full_content = self._merge_into_file(memory_path, section_content)
+            section_content = _build_section(memory_recs)
+            full_content = _merge_into_file(memory_path, section_content)
             result.add(memory_path, full_content)
-
             if not dry_run:
                 memory_path.parent.mkdir(parents=True, exist_ok=True)
                 memory_path.write_text(full_content)
@@ -319,47 +138,15 @@ class ClaudeCodeWriter(ContextWriter):
         return result
 
     def _resolve_context_path(self, project: ProjectInfo) -> Path:
-        """Resolve path for CLAUDE.md."""
         if project.context_file:
             return project.context_file
         return project.project_path / "CLAUDE.md"
 
     def _resolve_memory_path(self, project: ProjectInfo) -> Path:
-        """Resolve path for MEMORY.md."""
         if project.memory_file:
             return project.memory_file
         return project.data_path / "memory" / "MEMORY.md"
 
-    def _build_section(self, recommendations: list[Recommendation]) -> str:
-        """Build the marker-delimited section content."""
-        now = datetime.now(timezone.utc).strftime("%Y-%m-%d")
-        lines = [
-            _MARKER_START,
-            "## Headroom Learned Patterns",
-            f"*Auto-generated by `headroom learn` on {now} — do not edit manually*",
-            "",
-        ]
-
-        for rec in recommendations:
-            lines.append(f"### {rec.section}")
-            lines.append(rec.content)
-            lines.append("")
-
-        lines.append(_MARKER_END)
-        return "\n".join(lines)
-
-    def _merge_into_file(self, file_path: Path, section: str) -> str:
-        """Merge the section into an existing file, replacing any prior section."""
-        if file_path.exists():
-            existing = file_path.read_text()
-            # Replace existing headroom section
-            if _MARKER_START in existing:
-                return _MARKER_PATTERN.sub(section, existing)
-            # Append to end
-            return existing.rstrip() + "\n\n" + section + "\n"
-        else:
-            return section + "\n"
-
 
 # =============================================================================
 # Codex Writer (OpenAI Codex CLI)
@@ -367,11 +154,7 @@ class ClaudeCodeWriter(ContextWriter):
 
 
 class CodexWriter(ContextWriter):
-    """Writes learned patterns to AGENTS.md and instructions.md for Codex CLI.
-
-    Codex reads AGENTS.md from project root and ~/.codex/ for instructions.
-    instructions.md in ~/.codex/ acts as persistent cross-session memory.
-    """
+    """Writes learned patterns to AGENTS.md and instructions.md for Codex CLI."""
 
     def write(
         self,
@@ -387,8 +170,8 @@ class CodexWriter(ContextWriter):
 
         if context_recs:
             agents_md = project.context_file or (project.project_path / "AGENTS.md")
-            section_content = self._build_section(context_recs)
-            full_content = self._merge_into_file(agents_md, section_content)
+            section_content = _build_section(context_recs)
+            full_content = _merge_into_file(agents_md, section_content)
             result.add(agents_md, full_content)
             if not dry_run:
                 agents_md.parent.mkdir(parents=True, exist_ok=True)
@@ -396,8 +179,8 @@ class CodexWriter(ContextWriter):
 
         if memory_recs:
             instructions_md = project.memory_file or (project.data_path.parent / "instructions.md")
-            section_content = self._build_section(memory_recs)
-            full_content = self._merge_into_file(instructions_md, section_content)
+            section_content = _build_section(memory_recs)
+            full_content = _merge_into_file(instructions_md, section_content)
             result.add(instructions_md, full_content)
             if not dry_run:
                 instructions_md.parent.mkdir(parents=True, exist_ok=True)
@@ -405,29 +188,6 @@ class CodexWriter(ContextWriter):
 
         return result
 
-    def _build_section(self, recommendations: list[Recommendation]) -> str:
-        now = datetime.now(timezone.utc).strftime("%Y-%m-%d")
-        lines = [
-            _MARKER_START,
-            "## Headroom Learned Patterns",
-            f"*Auto-generated by `headroom learn` on {now} — do not edit manually*",
-            "",
-        ]
-        for rec in recommendations:
-            lines.append(f"### {rec.section}")
-            lines.append(rec.content)
-            lines.append("")
-        lines.append(_MARKER_END)
-        return "\n".join(lines)
-
-    def _merge_into_file(self, file_path: Path, section: str) -> str:
-        if file_path.exists():
-            existing = file_path.read_text()
-            if _MARKER_START in existing:
-                return _MARKER_PATTERN.sub(section, existing)
-            return existing.rstrip() + "\n\n" + section + "\n"
-        return section + "\n"
-
 
 # =============================================================================
 # Gemini Writer (Google Gemini CLI)
@@ -435,11 +195,7 @@ class CodexWriter(ContextWriter):
 
 
 class GeminiWriter(ContextWriter):
-    """Writes learned patterns to GEMINI.md for Gemini CLI.
-
-    Gemini reads GEMINI.md from project root and ~/.gemini/ for persistent context.
-    No separate memory file — everything goes into GEMINI.md.
-    """
+    """Writes learned patterns to GEMINI.md for Gemini CLI."""
 
     def write(
         self,
@@ -450,40 +206,15 @@ class GeminiWriter(ContextWriter):
         result = WriteResult()
         result.dry_run = dry_run
 
-        # Gemini has no memory file — all recommendations go to GEMINI.md
-        all_recs = recommendations
-        if not all_recs:
+        if not recommendations:
             return result
 
         gemini_md = project.context_file or (project.project_path / "GEMINI.md")
-        section_content = self._build_section(all_recs)
-        full_content = self._merge_into_file(gemini_md, section_content)
+        section_content = _build_section(recommendations)
+        full_content = _merge_into_file(gemini_md, section_content)
         result.add(gemini_md, full_content)
         if not dry_run:
             gemini_md.parent.mkdir(parents=True, exist_ok=True)
             gemini_md.write_text(full_content)
 
         return result
-
-    def _build_section(self, recommendations: list[Recommendation]) -> str:
-        now = datetime.now(timezone.utc).strftime("%Y-%m-%d")
-        lines = [
-            _MARKER_START,
-            "## Headroom Learned Patterns",
-            f"*Auto-generated by `headroom learn` on {now} — do not edit manually*",
-            "",
-        ]
-        for rec in recommendations:
-            lines.append(f"### {rec.section}")
-            lines.append(rec.content)
-            lines.append("")
-        lines.append(_MARKER_END)
-        return "\n".join(lines)
-
-    def _merge_into_file(self, file_path: Path, section: str) -> str:
-        if file_path.exists():
-            existing = file_path.read_text()
-            if _MARKER_START in existing:
-                return _MARKER_PATTERN.sub(section, existing)
-            return existing.rstrip() + "\n\n" + section + "\n"
-        return section + "\n"

headroom/perf/__init__.py

@@ -0,0 +1,4 @@
+"""Headroom performance analysis.
+
+Parse proxy logs and surface actionable insights via `headroom perf`.
+"""

headroom/perf/analyzer.py

@@ -0,0 +1,461 @@
+"""Analyze headroom proxy logs for performance insights.
+
+Parses PERF log lines from ~/.headroom/logs/proxy.log* and produces
+actionable reports on token savings, cache efficiency, and transform impact.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+
+LOG_DIR = Path.home() / ".headroom" / "logs"
+
+# Matches: 2026-03-07 13:38:31,009 - headroom.proxy - INFO - [hr_...] PERF model=... ...
+_PERF_RE = re.compile(
+    r"^(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+) .* \[(?P<rid>[^\]]+)\] PERF (?P<kv>.+)$"
+)
+
+# Matches: content_router: 51 msgs — ...
+_ROUTER_RE = re.compile(r"content_router: (?P<msgs>\d+) msgs — (?P<detail>.+)$")
+
+# Matches: Transform content_router: 52503 -> 26006 tokens (saved 26497)
+_TRANSFORM_RE = re.compile(
+    r"Transform (?P<name>\w+): (?P<before>\d+) -> (?P<after>\d+) tokens \(saved (?P<saved>\d+)\)"
+)
+
+# Matches: Pipeline complete: 52503 -> 26006 tokens (saved 26497, 50.5% reduction)
+_PIPELINE_RE = re.compile(
+    r"Pipeline complete: (?P<before>\d+) -> (?P<after>\d+) tokens "
+    r"\(saved (?P<saved>\d+), (?P<pct>[\d.]+)% reduction\)"
+)
+
+# Matches: TOIN: 105 patterns, 3837 compressions, 0 retrievals, 0.0% retrieval rate
+_TOIN_RE = re.compile(
+    r"TOIN: (?P<patterns>\d+) patterns, (?P<compressions>\d+) compressions, "
+    r"(?P<retrievals>\d+) retrievals, (?P<rate>[\d.]+)% retrieval rate"
+)
+
+
+def _parse_kv(kv_str: str) -> dict[str, str]:
+    """Parse key=value pairs from a PERF log line."""
+    result: dict[str, str] = {}
+    for part in kv_str.split():
+        if "=" in part:
+            k, v = part.split("=", 1)
+            result[k] = v
+    return result
+
+
+@dataclass
+class PerfRecord:
+    """A single parsed PERF log entry."""
+
+    timestamp: str
+    request_id: str
+    model: str = ""
+    num_messages: int = 0
+    tokens_before: int = 0
+    tokens_after: int = 0
+    tokens_saved: int = 0
+    cache_read: int = 0
+    cache_write: int = 0
+    cache_hit_pct: int = 0
+    optimization_ms: float = 0
+    transforms: list[str] = field(default_factory=list)
+
+
+@dataclass
+class RouterRecord:
+    """A parsed content_router summary line."""
+
+    timestamp: str
+    num_messages: int = 0
+    compressed: int = 0
+    excluded: int = 0
+    skipped: int = 0
+    unchanged: int = 0
+    content_blocks: int = 0
+
+
+@dataclass
+class TransformRecord:
+    """A parsed per-transform line."""
+
+    timestamp: str
+    name: str = ""
+    tokens_before: int = 0
+    tokens_after: int = 0
+    tokens_saved: int = 0
+
+
+@dataclass
+class ToinRecord:
+    """A parsed TOIN status line."""
+
+    timestamp: str
+    patterns: int = 0
+    compressions: int = 0
+    retrievals: int = 0
+    retrieval_rate: float = 0.0
+
+
+@dataclass
+class PerfReport:
+    """Aggregated performance report."""
+
+    perf_records: list[PerfRecord] = field(default_factory=list)
+    router_records: list[RouterRecord] = field(default_factory=list)
+    transform_records: list[TransformRecord] = field(default_factory=list)
+    toin_records: list[ToinRecord] = field(default_factory=list)
+    log_files_read: int = 0
+    total_lines_parsed: int = 0
+
+
+def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
+    """Parse all proxy log files and return structured records.
+
+    Args:
+        last_n_hours: Only include records from the last N hours (default 7 days).
+
+    Returns:
+        PerfReport with all parsed records.
+    """
+    report = PerfReport()
+
+    if not LOG_DIR.exists():
+        return report
+
+    # Collect log files: proxy.log, proxy.log.1, proxy.log.2, ...
+    log_files = sorted(LOG_DIR.glob("proxy.log*"), key=lambda p: p.stat().st_mtime)
+
+    for log_file in log_files:
+        report.log_files_read += 1
+        try:
+            with open(log_file, encoding="utf-8", errors="replace") as f:
+                for line in f:
+                    report.total_lines_parsed += 1
+                    line = line.rstrip()
+
+                    # PERF lines (richest data)
+                    m = _PERF_RE.match(line)
+                    if m:
+                        kv = _parse_kv(m.group("kv"))
+                        transforms_str = kv.get("transforms", "none")
+                        transforms = transforms_str.split(",") if transforms_str != "none" else []
+                        report.perf_records.append(
+                            PerfRecord(
+                                timestamp=m.group("ts"),
+                                request_id=m.group("rid"),
+                                model=kv.get("model", ""),
+                                num_messages=int(kv.get("msgs", 0)),
+                                tokens_before=int(kv.get("tok_before", 0)),
+                                tokens_after=int(kv.get("tok_after", 0)),
+                                tokens_saved=int(kv.get("tok_saved", 0)),
+                                cache_read=int(kv.get("cache_read", 0)),
+                                cache_write=int(kv.get("cache_write", 0)),
+                                cache_hit_pct=int(kv.get("cache_hit_pct", 0)),
+                                optimization_ms=float(kv.get("opt_ms", 0)),
+                                transforms=transforms,
+                            )
+                        )
+                        continue
+
+                    # content_router summary lines
+                    if "content_router:" in line and "msgs" in line:
+                        m2 = _ROUTER_RE.search(line)
+                        if m2:
+                            ts = line[:23]
+                            detail = m2.group("detail")
+                            rec = RouterRecord(
+                                timestamp=ts,
+                                num_messages=int(m2.group("msgs")),
+                            )
+                            # Parse counts from detail string
+                            for part in detail.split(","):
+                                part = part.strip()
+                                num_match = re.match(r"(\d+)\s+(\w+)", part)
+                                if num_match:
+                                    count = int(num_match.group(1))
+                                    kind = num_match.group(2)
+                                    if kind == "compressed":
+                                        rec.compressed = count
+                                    elif kind == "excluded":
+                                        rec.excluded = count
+                                    elif kind == "skipped":
+                                        rec.skipped = count
+                                    elif kind == "unchanged":
+                                        rec.unchanged = count
+                                    elif kind == "content" and "block" in part:
+                                        rec.content_blocks = count
+                            report.router_records.append(rec)
+                            continue
+
+                    # Per-transform lines
+                    m3 = _TRANSFORM_RE.search(line)
+                    if m3:
+                        ts = line[:23]
+                        report.transform_records.append(
+                            TransformRecord(
+                                timestamp=ts,
+                                name=m3.group("name"),
+                                tokens_before=int(m3.group("before")),
+                                tokens_after=int(m3.group("after")),
+                                tokens_saved=int(m3.group("saved")),
+                            )
+                        )
+                        continue
+
+                    # TOIN status lines
+                    m4 = _TOIN_RE.search(line)
+                    if m4:
+                        ts = line[:23]
+                        report.toin_records.append(
+                            ToinRecord(
+                                timestamp=ts,
+                                patterns=int(m4.group("patterns")),
+                                compressions=int(m4.group("compressions")),
+                                retrievals=int(m4.group("retrievals")),
+                                retrieval_rate=float(m4.group("rate")),
+                            )
+                        )
+
+        except OSError:
+            continue
+
+    return report
+
+
+def format_report(report: PerfReport) -> str:
+    """Format a PerfReport into a human-readable string."""
+    lines: list[str] = []
+
+    if not report.perf_records and not report.router_records:
+        lines.append("No performance data found in ~/.headroom/logs/")
+        lines.append("")
+        lines.append("Start the proxy to begin collecting data:")
+        lines.append("  headroom proxy")
+        return "\n".join(lines)
+
+    # Header
+    lines.append("Headroom Performance Report")
+    lines.append("=" * 60)
+    lines.append("")
+
+    records = report.perf_records
+
+    if records:
+        # Overview
+        total_before = sum(r.tokens_before for r in records)
+        total_after = sum(r.tokens_after for r in records)
+        total_saved = sum(r.tokens_saved for r in records)
+        pct = (total_saved / total_before * 100) if total_before > 0 else 0
+
+        models = {r.model for r in records}
+        lines.append(f"Requests:     {len(records)}")
+        lines.append(f"Models:       {', '.join(sorted(models))}")
+        lines.append(
+            f"Tokens:       {total_before:,} input -> {total_after:,} after transforms "
+            f"({pct:.1f}% reduction)"
+        )
+        lines.append(f"Total saved:  {total_saved:,} tokens")
+        lines.append("")
+
+        # Cache analysis
+        cache_records = [r for r in records if (r.cache_read + r.cache_write) > 0]
+        if cache_records:
+            lines.append("Cache Performance")
+            lines.append("-" * 40)
+            total_cr = sum(r.cache_read for r in cache_records)
+            total_cw = sum(r.cache_write for r in cache_records)
+            total_cache = total_cr + total_cw
+            hit_pct = (total_cr / total_cache * 100) if total_cache > 0 else 0
+            lines.append(f"  Cache read:    {total_cr:,} tokens")
+            lines.append(f"  Cache write:   {total_cw:,} tokens")
+            lines.append(f"  Hit rate:      {hit_pct:.1f}%")
+
+            # Identify cache instability: requests where write >> read
+            unstable = [r for r in cache_records if r.cache_write > r.cache_read * 2]
+            if unstable:
+                lines.append(
+                    f"  Unstable:      {len(unstable)}/{len(cache_records)} requests "
+                    f"had cache_write > 2x cache_read"
+                )
+
+            # Show cache progression (first 5 vs last 5)
+            if len(cache_records) >= 10:
+                first5_cr = sum(r.cache_read for r in cache_records[:5])
+                first5_cw = sum(r.cache_write for r in cache_records[:5])
+                last5_cr = sum(r.cache_read for r in cache_records[-5:])
+                last5_cw = sum(r.cache_write for r in cache_records[-5:])
+                lines.append(f"  First 5 avg:   read={first5_cr // 5:,} write={first5_cw // 5:,}")
+                lines.append(f"  Last 5 avg:    read={last5_cr // 5:,} write={last5_cw // 5:,}")
+                if last5_cr > first5_cr * 2:
+                    lines.append("  -> Cache stabilizing over conversation lifetime")
+                elif first5_cw > first5_cr * 3:
+                    lines.append(
+                        "  ! Early turns have poor cache hits — "
+                        "compression decisions may be flipping"
+                    )
+            lines.append("")
+
+        # Optimization latency
+        opt_times = [r.optimization_ms for r in records if r.optimization_ms > 0]
+        if opt_times:
+            avg_opt = sum(opt_times) / len(opt_times)
+            max_opt = max(opt_times)
+            lines.append("Optimization Overhead")
+            lines.append("-" * 40)
+            lines.append(f"  Average:  {avg_opt:.0f}ms")
+            lines.append(f"  Max:      {max_opt:.0f}ms")
+            slow = [t for t in opt_times if t > 500]
+            if slow:
+                lines.append(f"  >500ms:   {len(slow)} requests")
+            lines.append("")
+
+        # Conversation size distribution
+        msg_counts = [r.num_messages for r in records if r.num_messages > 0]
+        if msg_counts:
+            lines.append("Conversation Size")
+            lines.append("-" * 40)
+            lines.append(f"  Min msgs:  {min(msg_counts)}")
+            lines.append(f"  Max msgs:  {max(msg_counts)}")
+            lines.append(f"  Avg msgs:  {sum(msg_counts) // len(msg_counts)}")
+            lines.append("")
+
+    # Transform effectiveness (from transform_records)
+    if report.transform_records:
+        lines.append("Transform Effectiveness")
+        lines.append("-" * 40)
+        by_name: dict[str, list[TransformRecord]] = {}
+        for tr in report.transform_records:
+            by_name.setdefault(tr.name, []).append(tr)
+        for name, recs in sorted(by_name.items(), key=lambda x: -sum(r.tokens_saved for r in x[1])):
+            total_s = sum(r.tokens_saved for r in recs)
+            total_b = sum(r.tokens_before for r in recs)
+            avg_pct = (total_s / total_b * 100) if total_b > 0 else 0
+            lines.append(
+                f"  {name}: {avg_pct:.1f}% avg reduction, {len(recs)} uses, {total_s:,} saved"
+            )
+        lines.append("")
+
+    # Router routing breakdown
+    if report.router_records:
+        lines.append("Content Router Routing")
+        lines.append("-" * 40)
+        total_compressed = sum(r.compressed for r in report.router_records)
+        total_excluded = sum(r.excluded for r in report.router_records)
+        total_skipped = sum(r.skipped for r in report.router_records)
+        total_unchanged = sum(r.unchanged for r in report.router_records)
+        total_all = total_compressed + total_excluded + total_skipped + total_unchanged
+        if total_all > 0:
+            lines.append(
+                f"  Compressed:  {total_compressed} ({total_compressed / total_all * 100:.0f}%)"
+            )
+            lines.append(
+                f"  Excluded:    {total_excluded} ({total_excluded / total_all * 100:.0f}%) — Read/Glob outputs"
+            )
+            lines.append(
+                f"  Skipped:     {total_skipped} ({total_skipped / total_all * 100:.0f}%) — <50 words"
+            )
+            lines.append(
+                f"  Unchanged:   {total_unchanged} ({total_unchanged / total_all * 100:.0f}%) — ratio too high"
+            )
+        if total_excluded > total_compressed * 3:
+            lines.append("  ! Excluded tools dominate — consider compressing stale Read outputs")
+        lines.append("")
+
+    # TOIN status
+    if report.toin_records:
+        latest = report.toin_records[-1]
+        lines.append("TOIN Learning")
+        lines.append("-" * 40)
+        lines.append(f"  Patterns:     {latest.patterns}")
+        lines.append(f"  Compressions: {latest.compressions:,}")
+        lines.append(f"  Retrievals:   {latest.retrievals} ({latest.retrieval_rate}%)")
+        if latest.retrieval_rate == 0 and latest.compressions > 100:
+            lines.append("  ! 0% retrieval rate — TOIN learning but never used")
+        lines.append("")
+
+    # Recommendations
+    recommendations = _generate_recommendations(report)
+    if recommendations:
+        lines.append("Recommendations")
+        lines.append("-" * 40)
+        for i, rec in enumerate(recommendations, 1):
+            lines.append(f"  {i}. {rec}")
+        lines.append("")
+
+    # Footer
+    lines.append(
+        f"Log files: {report.log_files_read} | Lines parsed: {report.total_lines_parsed:,}"
+    )
+    lines.append(f"Log dir: {LOG_DIR}")
+
+    return "\n".join(lines)
+
+
+def _generate_recommendations(report: PerfReport) -> list[str]:
+    """Generate actionable recommendations from the report data."""
+    recs: list[str] = []
+
+    if report.perf_records:
+        cache_recs = [r for r in report.perf_records if (r.cache_read + r.cache_write) > 0]
+        if cache_recs:
+            total_cr = sum(r.cache_read for r in cache_recs)
+            total_cw = sum(r.cache_write for r in cache_recs)
+            if total_cw > total_cr * 1.5:
+                recs.append(
+                    "Cache prefix unstable — compression decisions may be flipping "
+                    "across turns due to adaptive min_ratio threshold"
+                )
+
+            # Check early-turn instability
+            if len(cache_recs) >= 5:
+                first5 = cache_recs[:5]
+                early_ratio = sum(r.cache_read for r in first5) / max(
+                    1, sum(r.cache_write for r in first5)
+                )
+                if early_ratio < 0.5:
+                    recs.append(
+                        "First 5 turns have very low cache hit ratio — "
+                        "consider pinning compression decisions for prefix stability"
+                    )
+
+        # Optimization latency
+        slow = [r for r in report.perf_records if r.optimization_ms > 500]
+        if len(slow) > len(report.perf_records) * 0.2:
+            recs.append(
+                f"{len(slow)} requests took >500ms for optimization — "
+                "consider disabling LLMLingua or reducing transform pipeline"
+            )
+
+    if report.router_records:
+        total_excluded = sum(r.excluded for r in report.router_records)
+        total_compressed = sum(r.compressed for r in report.router_records)
+        if total_excluded > 0 and total_compressed > 0:
+            if total_excluded > total_compressed * 3:
+                recs.append(
+                    "Read/Glob outputs are majority of messages but excluded — "
+                    "compress stale reads (>10 turns old) for significant savings"
+                )
+
+    if report.toin_records:
+        latest = report.toin_records[-1]
+        if latest.retrieval_rate == 0 and latest.compressions > 100:
+            recs.append(
+                "TOIN has 0% retrieval rate with "
+                f"{latest.compressions:,} compressions — review CCR integration"
+            )
+
+    # Check cache_aligner effectiveness from transform records
+    for tr in report.transform_records:
+        if tr.name == "cache_aligner" and tr.tokens_saved < 10:
+            recs.append(
+                "cache_aligner saving <10 tokens — "
+                "consider disabling (system prompt likely has no dynamic content)"
+            )
+            break
+
+    return recs

headroom/proxy/server.py

@@ -140,6 +140,41 @@ logging.basicConfig(
 )
 logger = logging.getLogger("headroom.proxy")
 
+# Always-on file logging to ~/.headroom/logs/ for `headroom perf` analysis
+_HEADROOM_LOG_DIR = Path.home() / ".headroom" / "logs"
+
+
+def _setup_file_logging() -> None:
+    """Add a RotatingFileHandler to the headroom root logger.
+
+    Writes to ~/.headroom/logs/proxy.log with automatic rotation:
+    - Rotates at 10 MB
+    - Keeps 5 backups (~50 MB max)
+    """
+    from logging.handlers import RotatingFileHandler
+
+    try:
+        _HEADROOM_LOG_DIR.mkdir(parents=True, exist_ok=True)
+        log_path = _HEADROOM_LOG_DIR / "proxy.log"
+        handler = RotatingFileHandler(
+            log_path,
+            maxBytes=10 * 1024 * 1024,  # 10 MB
+            backupCount=5,
+            encoding="utf-8",
+        )
+        handler.setLevel(logging.INFO)
+        handler.setFormatter(
+            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+        )
+        # Attach to the headroom root logger so all sub-loggers are captured
+        logging.getLogger("headroom").addHandler(handler)
+    except OSError:
+        # Non-fatal: can't write logs (read-only fs, permissions, etc.)
+        pass
+
+
+_setup_file_logging()
+
 # Maximum request body size (100MB - increased to support image-heavy requests)
 MAX_REQUEST_BODY_SIZE = 100 * 1024 * 1024
 
@@ -2055,8 +2090,7 @@ class HeadroomProxy:
                             except Exception:
                                 pass
                             logger.error(
-                                f"CCR: API call failed: {e}, "
-                                f"response headers: {resp_headers}"
+                                f"CCR: API call failed: {e}, response headers: {resp_headers}"
                             )
                             raise
 
@@ -2080,7 +2114,9 @@ class HeadroomProxy:
                         try:
                             ccr_content = json.dumps(final_resp_json).encode()
                         except (TypeError, ValueError) as json_err:
-                            logger.warning(f"[{request_id}] CCR: JSON serialization failed: {json_err}")
+                            logger.warning(
+                                f"[{request_id}] CCR: JSON serialization failed: {json_err}"
+                            )
                             ccr_content = json.dumps(resp_json).encode()
                         response = httpx.Response(
                             status_code=200,
@@ -2237,15 +2273,21 @@ class HeadroomProxy:
                         )
                     )
 
-                # Log to console
-                if tokens_saved > 0:
-                    logger.info(
-                        f"[{request_id}] {model}: {original_tokens:,} → {optimized_tokens:,} "
-                        f"(saved {tokens_saved:,} tokens, ${savings_usd:.4f})"
-                        if savings_usd
-                        else f"[{request_id}] {model}: {original_tokens:,} → {optimized_tokens:,} "
-                        f"(saved {tokens_saved:,} tokens)"
-                    )
+                # Structured perf log line for `headroom perf` analysis
+                num_msgs = len(messages)
+                resp_usage = resp_json.get("usage", {}) if resp_json else {}
+                cr = resp_usage.get("cache_read_input_tokens", 0)
+                cw = resp_usage.get("cache_creation_input_tokens", 0)
+                chp = round(cr / (cr + cw) * 100) if (cr + cw) > 0 else 0
+                logger.info(
+                    f"[{request_id}] PERF "
+                    f"model={model} msgs={num_msgs} "
+                    f"tok_before={original_tokens} tok_after={optimized_tokens} "
+                    f"tok_saved={tokens_saved} "
+                    f"cache_read={cr} cache_write={cw} cache_hit_pct={chp} "
+                    f"opt_ms={optimization_latency:.0f} "
+                    f"transforms={','.join(transforms_applied) if transforms_applied else 'none'}"
+                )
 
                 # Remove compression headers since httpx already decompressed the response
                 response_headers = dict(response.headers)
@@ -3852,16 +3894,23 @@ class HeadroomProxy:
                 cache_read_tokens = stream_state["cache_read_input_tokens"]
                 cache_write_tokens = stream_state["cache_creation_input_tokens"]
 
-                # INFO logging for token tracking (temporary for debugging)
-                if api_input_tokens is None:
-                    logger.info(
-                        f"[{request_id}] No input_tokens from API, using estimate: {optimized_tokens}"
-                    )
-                else:
-                    logger.info(
-                        f"[{request_id}] Final tokens: input={api_input_tokens}, "
-                        f"cache_read={cache_read_tokens}, cache_write={cache_write_tokens}"
-                    )
+                # Structured perf log line for `headroom perf` analysis
+                num_msgs = len(body.get("messages", []))
+                cache_hit_pct = (
+                    round(cache_read_tokens / (cache_read_tokens + cache_write_tokens) * 100)
+                    if (cache_read_tokens + cache_write_tokens) > 0
+                    else 0
+                )
+                logger.info(
+                    f"[{request_id}] PERF "
+                    f"model={model} msgs={num_msgs} "
+                    f"tok_before={original_tokens} tok_after={optimized_tokens} "
+                    f"tok_saved={tokens_saved} "
+                    f"cache_read={cache_read_tokens} cache_write={cache_write_tokens} "
+                    f"cache_hit_pct={cache_hit_pct} "
+                    f"opt_ms={optimization_latency:.0f} "
+                    f"transforms={','.join(transforms_applied) if transforms_applied else 'none'}"
+                )
 
                 # Normalize input tokens based on provider semantics:
                 # - Anthropic: input_tokens excludes cache_read (it's separate), pass as-is
@@ -3913,11 +3962,6 @@ class HeadroomProxy:
                     savings_usd=savings_usd or 0,
                 )
 
-                if tokens_saved > 0:
-                    logger.info(
-                        f"[{request_id}] {model}: saved {tokens_saved:,} tokens (streaming)"
-                    )
-
         return StreamingResponse(
             generate(),
             media_type="text/event-stream",
@@ -4045,10 +4089,17 @@ class HeadroomProxy:
                         )
                     )
 
-                if tokens_saved > 0:
-                    logger.info(
-                        f"[{request_id}] Bedrock {model}: saved {tokens_saved:,} tokens (streaming)"
-                    )
+                # Structured perf log line for `headroom perf` analysis
+                num_msgs = len(body.get("messages", []))
+                logger.info(
+                    f"[{request_id}] PERF "
+                    f"model={model} msgs={num_msgs} "
+                    f"tok_before={original_tokens} tok_after={optimized_tokens} "
+                    f"tok_saved={tokens_saved} "
+                    f"cache_read=0 cache_write=0 cache_hit_pct=0 "
+                    f"opt_ms={optimization_latency:.0f} "
+                    f"transforms={','.join(transforms_applied) if transforms_applied else 'none'}"
+                )
 
         return StreamingResponse(
             generate(),

tests/test_learn/test_analyzer.py

@@ -1,12 +1,21 @@
-"""Tests for failure analyzers — generic tool call pattern recognition."""
+"""Tests for session analyzer — digest builder and LLM-based analysis."""
 
 from pathlib import Path
+from unittest.mock import MagicMock, patch
 
-from headroom.learn.analyzer import FailureAnalyzer
+from headroom.learn.analyzer import (
+    SessionAnalyzer,
+    _build_digest,
+    _detect_default_model,
+    _parse_llm_response,
+)
 from headroom.learn.models import (
+    AnalysisResult,
     ErrorCategory,
     ProjectInfo,
+    RecommendationTarget,
     SessionData,
+    SessionEvent,
     ToolCall,
 )
 
@@ -26,6 +35,7 @@ def _tc(
     is_error: bool = False,
     error_category: ErrorCategory = ErrorCategory.UNKNOWN,
     msg_index: int = 0,
+    output_bytes: int = 0,
 ) -> ToolCall:
     return ToolCall(
         name=name,
@@ -35,314 +45,330 @@ def _tc(
         is_error=is_error,
         error_category=error_category,
         msg_index=msg_index,
-        output_bytes=len(output),
+        output_bytes=output_bytes or len(output),
     )
 
 
-class TestAnalyzerBasics:
-    def test_empty_sessions(self):
-        analyzer = FailureAnalyzer()
-        report = analyzer.analyze(_project(), [])
-        assert report.total_calls == 0
-        assert report.total_failures == 0
-        assert report.failure_rate == 0.0
+# =============================================================================
+# Digest Builder Tests
+# =============================================================================
 
-    def test_no_failures(self):
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id="s1",
-                tool_calls=[_tc(msg_index=i) for i in range(10)],
-            )
-        ]
-        report = analyzer.analyze(_project(), sessions)
-        assert report.total_calls == 10
-        assert report.total_failures == 0
 
-    def test_basic_failure_counting(self):
-        analyzer = FailureAnalyzer()
+class TestDigestBuilder:
+    def test_includes_project_info(self):
+        project = _project()
+        sessions = [SessionData(session_id="s1", tool_calls=[_tc()])]
+        digest = _build_digest(project, sessions)
+        assert "test-project" in digest
+        assert "/tmp/test-project" in digest
+
+    def test_includes_session_stats(self):
         sessions = [
             SessionData(
-                session_id="s1",
-                tool_calls=[
-                    _tc(msg_index=0),
-                    _tc(msg_index=1, is_error=True, output="Error: something broke"),
-                    _tc(msg_index=2),
-                ],
+                session_id="abc123",
+                tool_calls=[_tc(msg_index=0), _tc(msg_index=1, is_error=True, output="Error!")],
             )
         ]
-        report = analyzer.analyze(_project(), sessions)
-        assert report.total_calls == 3
-        assert report.total_failures == 1
-
+        digest = _build_digest(_project(), sessions)
+        assert "abc123" in digest
+        assert "2 calls" in digest
+        assert "1 failure" in digest
 
-class TestEnvironmentAnalyzer:
-    def test_detects_wrong_python(self):
-        """Module not found with python3 + successes with uv run → learn correct command."""
-        analyzer = FailureAnalyzer()
+    def test_includes_tool_call_details(self):
         sessions = [
             SessionData(
                 session_id="s1",
                 tool_calls=[
-                    # Failures with python3
                     _tc(
-                        name="Bash",
-                        input_data={"command": "python3 -c 'import mylib'"},
-                        output="ModuleNotFoundError: No module named 'mylib'",
-                        is_error=True,
-                        error_category=ErrorCategory.MODULE_NOT_FOUND,
+                        name="Read",
+                        input_data={"file_path": "/src/foo.py"},
+                        output="contents",
                         msg_index=0,
                     ),
                     _tc(
                         name="Bash",
-                        input_data={"command": "python3 -c 'import mylib'"},
+                        input_data={"command": "python3 run.py"},
                         output="ModuleNotFoundError",
                         is_error=True,
                         error_category=ErrorCategory.MODULE_NOT_FOUND,
                         msg_index=1,
                     ),
-                    # Success with uv run
-                    _tc(
-                        name="Bash",
-                        input_data={"command": "uv run python -c 'import mylib'"},
-                        output="ok",
-                        msg_index=2,
-                    ),
-                    _tc(
-                        name="Bash",
-                        input_data={"command": "uv run python -c 'import mylib'"},
-                        output="ok",
-                        msg_index=3,
-                    ),
-                    _tc(
-                        name="Bash",
-                        input_data={"command": "uv run python -c 'import mylib'"},
-                        output="ok",
-                        msg_index=4,
-                    ),
                 ],
             )
         ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.environment_facts) >= 1
-        fact = report.environment_facts[0]
-        assert fact.category == "python"
-        assert "uv run" in fact.correct_command
-        assert "python3" in fact.wrong_commands
+        digest = _build_digest(_project(), sessions)
+        assert "/src/foo.py" in digest
+        assert "python3 run.py" in digest
+        assert "ERROR" in digest
+        assert "ModuleNotFoundError" in digest
 
+    def test_includes_user_messages(self):
+        tc = _tc(msg_index=0)
+        events = [
+            SessionEvent(type="tool_call", msg_index=0, tool_call=tc),
+            SessionEvent(type="user_message", msg_index=1, text="Use uv run instead"),
+        ]
+        sessions = [SessionData(session_id="s1", tool_calls=[tc], events=events)]
+        digest = _build_digest(_project(), sessions)
+        assert "USER:" in digest
+        assert "Use uv run instead" in digest
 
-class TestStructureAnalyzer:
-    def test_detects_missing_paths(self):
-        """Files that repeatedly fail Read → learned as missing."""
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id="s1",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/src/missing.py"},
-                        output="No such file",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_NOT_FOUND,
-                        msg_index=0,
-                    ),
-                ],
-            ),
-            SessionData(
-                session_id="s2",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/src/missing.py"},
-                        output="No such file",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_NOT_FOUND,
-                        msg_index=0,
-                    ),
-                ],
+    def test_includes_subagent_summaries(self):
+        events = [
+            SessionEvent(
+                type="agent_summary",
+                msg_index=0,
+                agent_tool_count=150,
+                agent_tokens=60000,
+                agent_prompt="Explore all test files",
             ),
         ]
-        report = analyzer.analyze(_project(), sessions)
-        missing = [n for n in report.structure_notes if n.category == "missing_path"]
-        assert len(missing) >= 1
-        assert "/src/missing.py" in missing[0].path
+        sessions = [SessionData(session_id="s1", events=events)]
+        digest = _build_digest(_project(), sessions)
+        assert "SUBAGENT" in digest
+        assert "150 tool calls" in digest
+        assert "Explore all test files" in digest
 
-    def test_detects_large_files(self):
-        """Files that repeatedly trigger too-large errors → learned."""
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id="s1",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/src/huge.py"},
-                        output="file is too large",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_TOO_LARGE,
-                        msg_index=0,
-                    ),
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/src/huge.py"},
-                        output="file is too large",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_TOO_LARGE,
-                        msg_index=1,
-                    ),
-                ],
+    def test_includes_interruptions(self):
+        events = [
+            SessionEvent(
+                type="interruption",
+                msg_index=0,
+                text="[Request interrupted by user]",
             ),
         ]
-        report = analyzer.analyze(_project(), sessions)
-        large = [n for n in report.structure_notes if n.category == "large_file"]
-        assert len(large) >= 1
-        assert "/src/huge.py" in large[0].path
+        sessions = [SessionData(session_id="s1", events=events)]
+        digest = _build_digest(_project(), sessions)
+        assert "INTERRUPTED" in digest
 
-    def test_single_occurrence_not_reported(self):
-        """A single file_not_found shouldn't be reported (might be transient)."""
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id="s1",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/src/one_time.py"},
-                        output="No such file",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_NOT_FOUND,
-                        msg_index=0,
-                    ),
-                ],
-            )
-        ]
-        report = analyzer.analyze(_project(), sessions)
-        missing = [n for n in report.structure_notes if n.category == "missing_path"]
-        assert len(missing) == 0
+    def test_empty_sessions(self):
+        digest = _build_digest(_project(), [])
+        assert "0 sessions" in digest or "test-project" in digest
 
 
-class TestRetryAnalyzer:
-    def test_detects_stubborn_retries(self):
-        """Same tool failing 3+ times in a row → retry pattern."""
-        analyzer = FailureAnalyzer()
+# =============================================================================
+# LLM Response Parser Tests
+# =============================================================================
+
+
+class TestLLMResponseParser:
+    def test_parses_context_file_rules(self):
+        raw = {
+            "context_file_rules": [
+                {
+                    "section": "Environment",
+                    "content": "- Use `uv run python` instead of `python3`",
+                    "estimated_tokens_saved": 800,
+                    "evidence_count": 5,
+                }
+            ],
+            "memory_file_rules": [],
+        }
+        recs = _parse_llm_response(raw)
+        assert len(recs) == 1
+        assert recs[0].target == RecommendationTarget.CONTEXT_FILE
+        assert recs[0].section == "Environment"
+        assert "uv run python" in recs[0].content
+        assert recs[0].estimated_tokens_saved == 800
+        assert recs[0].evidence_count == 5
+
+    def test_parses_memory_file_rules(self):
+        raw = {
+            "context_file_rules": [],
+            "memory_file_rules": [
+                {
+                    "section": "User Preferences",
+                    "content": "- Do not auto-execute curl commands",
+                    "estimated_tokens_saved": 500,
+                    "evidence_count": 3,
+                }
+            ],
+        }
+        recs = _parse_llm_response(raw)
+        assert len(recs) == 1
+        assert recs[0].target == RecommendationTarget.MEMORY_FILE
+        assert "curl" in recs[0].content
+
+    def test_sorts_by_token_savings(self):
+        raw = {
+            "context_file_rules": [
+                {
+                    "section": "Paths",
+                    "content": "- Use correct paths",
+                    "estimated_tokens_saved": 200,
+                    "evidence_count": 2,
+                },
+                {
+                    "section": "Environment",
+                    "content": "- Use uv",
+                    "estimated_tokens_saved": 1000,
+                    "evidence_count": 5,
+                },
+            ],
+            "memory_file_rules": [],
+        }
+        recs = _parse_llm_response(raw)
+        assert recs[0].estimated_tokens_saved == 1000
+        assert recs[1].estimated_tokens_saved == 200
+
+    def test_handles_missing_fields(self):
+        raw = {
+            "context_file_rules": [
+                {"section": "Env", "content": "- stuff"},
+                {"section": "", "content": ""},  # should be skipped
+                {"not_a_real_field": True},  # should be skipped
+            ],
+            "memory_file_rules": [],
+        }
+        recs = _parse_llm_response(raw)
+        assert len(recs) == 1
+
+    def test_handles_empty_response(self):
+        recs = _parse_llm_response({})
+        assert recs == []
+
+    def test_handles_non_dict_entries(self):
+        raw = {"context_file_rules": ["not a dict", 42], "memory_file_rules": []}
+        recs = _parse_llm_response(raw)
+        assert recs == []
+
+
+# =============================================================================
+# Full Analyzer Integration Tests (mocked LLM)
+# =============================================================================
+
+
+class TestSessionAnalyzer:
+    def test_empty_sessions_no_llm_call(self):
+        """No failures + no events → no LLM call, empty result."""
+        analyzer = SessionAnalyzer()
+        result = analyzer.analyze(_project(), [])
+        assert result.total_calls == 0
+        assert result.total_failures == 0
+        assert result.recommendations == []
+
+    @patch("headroom.learn.analyzer._call_llm")
+    def test_calls_llm_with_digest(self, mock_call_llm: MagicMock):
+        mock_call_llm.return_value = {
+            "context_file_rules": [
+                {
+                    "section": "Environment",
+                    "content": "- Use uv run python",
+                    "estimated_tokens_saved": 800,
+                    "evidence_count": 3,
+                }
+            ],
+            "memory_file_rules": [],
+        }
+
+        analyzer = SessionAnalyzer()
         sessions = [
             SessionData(
                 session_id="s1",
                 tool_calls=[
-                    _tc(
-                        name="Bash",
-                        input_data={"command": "mkdir -p /x"},
-                        output="auto-denied",
-                        is_error=True,
-                        error_category=ErrorCategory.PERMISSION_DENIED,
-                        msg_index=i,
-                    )
-                    for i in range(5)
+                    _tc(msg_index=0, is_error=True, output="ModuleNotFoundError"),
+                    _tc(msg_index=1),
                 ],
             )
         ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.retry_patterns) >= 1
-        pattern = report.retry_patterns[0]
-        assert pattern.tool_name == "Bash"
-        assert pattern.max_retries_seen >= 5
-
-    def test_two_failures_not_stubborn(self):
-        """Only 2 failures shouldn't trigger a retry pattern (threshold is 3)."""
-        analyzer = FailureAnalyzer()
+        result = analyzer.analyze(_project(), sessions)
+
+        mock_call_llm.assert_called_once()
+        assert result.total_calls == 2
+        assert result.total_failures == 1
+        assert len(result.recommendations) == 1
+        assert "uv run python" in result.recommendations[0].content
+
+    @patch("headroom.learn.analyzer._call_llm")
+    def test_handles_llm_failure_gracefully(self, mock_call_llm: MagicMock):
+        mock_call_llm.side_effect = RuntimeError("API key not set")
+
+        analyzer = SessionAnalyzer()
         sessions = [
             SessionData(
                 session_id="s1",
-                tool_calls=[
-                    _tc(
-                        name="Glob",
-                        output="No matches",
-                        is_error=True,
-                        error_category=ErrorCategory.NO_MATCHES,
-                        msg_index=0,
-                    ),
-                    _tc(
-                        name="Glob",
-                        output="No matches",
-                        is_error=True,
-                        error_category=ErrorCategory.NO_MATCHES,
-                        msg_index=1,
-                    ),
-                    _tc(name="Glob", output="found.py", msg_index=2),  # Success breaks streak
-                ],
+                tool_calls=[_tc(msg_index=0, is_error=True, output="error")],
             )
         ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.retry_patterns) == 0
+        result = analyzer.analyze(_project(), sessions)
 
+        # Stats should still work, just no recommendations
+        assert result.total_calls == 1
+        assert result.total_failures == 1
+        assert result.recommendations == []
 
-class TestCrossSessionAnalyzer:
-    def test_cross_session_pattern(self):
-        """Same failure in 3+ sessions → cross-session pattern."""
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id=f"s{i}",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/docs/RESEARCH.md"},
-                        output="No such file",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_NOT_FOUND,
-                        msg_index=0,
-                    ),
-                ],
-            )
-            for i in range(4)
-        ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.cross_session_patterns) >= 1
-        assert any("RESEARCH.md" in p for p in report.cross_session_patterns)
+    @patch("headroom.learn.analyzer._call_llm")
+    def test_passes_events_to_digest(self, mock_call_llm: MagicMock):
+        """User messages and subagent events should appear in the digest."""
+        mock_call_llm.return_value = {"context_file_rules": [], "memory_file_rules": []}
 
-    def test_two_sessions_not_enough(self):
-        """Only 2 sessions shouldn't trigger cross-session (threshold is 3)."""
-        analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id=f"s{i}",
-                tool_calls=[
-                    _tc(
-                        name="Read",
-                        input_data={"file_path": "/rare.py"},
-                        output="No such file",
-                        is_error=True,
-                        error_category=ErrorCategory.FILE_NOT_FOUND,
-                        msg_index=0,
-                    ),
-                ],
-            )
-            for i in range(2)
+        tc = _tc(msg_index=0, is_error=True, output="error")
+        events = [
+            SessionEvent(type="tool_call", msg_index=0, tool_call=tc),
+            SessionEvent(type="user_message", msg_index=1, text="use venv python"),
         ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.cross_session_patterns) == 0
+        sessions = [SessionData(session_id="s1", tool_calls=[tc], events=events)]
+
+        analyzer = SessionAnalyzer()
+        analyzer.analyze(_project(), sessions)
+
+        # Check that the digest passed to the LLM includes user message
+        call_args = mock_call_llm.call_args
+        digest = call_args[0][0]  # first positional arg
+        assert "use venv python" in digest
+
+
+# =============================================================================
+# Model Auto-Detection
+# =============================================================================
+
+
+class TestDetectDefaultModel:
+    def test_anthropic_key(self, monkeypatch):
+        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        assert _detect_default_model() == "claude-sonnet-4-6"
 
+    def test_openai_key(self, monkeypatch):
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        assert _detect_default_model() == "gpt-4o"
+
+    def test_gemini_key(self, monkeypatch):
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+        monkeypatch.setenv("GEMINI_API_KEY", "test")
+        assert _detect_default_model() == "gemini/gemini-2.0-flash"
+
+    def test_anthropic_preferred_over_openai(self, monkeypatch):
+        monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+        monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+        assert _detect_default_model() == "claude-sonnet-4-6"
+
+    def test_no_keys_raises(self, monkeypatch):
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        import pytest
+
+        with pytest.raises(RuntimeError, match="No LLM API key found"):
+            _detect_default_model()
+
+
+# =============================================================================
+# Legacy Compatibility
+# =============================================================================
+
+
+class TestFailureAnalyzerCompat:
+    @patch("headroom.learn.analyzer._call_llm")
+    def test_legacy_alias_works(self, mock_call_llm: MagicMock):
+        from headroom.learn.analyzer import FailureAnalyzer
+
+        mock_call_llm.return_value = {"context_file_rules": [], "memory_file_rules": []}
 
-class TestPermissionAnalyzer:
-    def test_detects_repeated_denials(self):
-        """Commands denied 3+ times → permission note."""
         analyzer = FailureAnalyzer()
-        sessions = [
-            SessionData(
-                session_id="s1",
-                tool_calls=[
-                    _tc(
-                        name="Bash",
-                        input_data={"command": "mkdir -p /x"},
-                        output="auto-denied",
-                        is_error=True,
-                        error_category=ErrorCategory.PERMISSION_DENIED,
-                        msg_index=i,
-                    )
-                    for i in range(4)
-                ],
-            )
-        ]
-        report = analyzer.analyze(_project(), sessions)
-        assert len(report.permission_issues) >= 1
+        result = analyzer.analyze(_project(), [])
+        assert isinstance(result, AnalysisResult)

tests/test_learn/test_integration.py

@@ -3,114 +3,36 @@
 These tests run against actual conversation data on the machine.
 They verify the full pipeline: scan → analyze → recommend → write.
 Tests are skipped if the required data directories don't exist.
+The LLM-based analyzer tests require ANTHROPIC_API_KEY.
 
 Key behaviors tested:
-- Quality gates prevent writing when evidence is weak
-- False positives don't generate recommendations
-- Real Codex/Claude Code sessions produce sensible output
+- Empty recommendations don't create files
+- Real Codex/Claude Code sessions can be scanned
 - Skip logic: no file writes when nothing meaningful is found
 - Idempotency: running twice produces same output
 """
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
-from headroom.learn.analyzer import FailureAnalyzer
+from headroom.learn.analyzer import SessionAnalyzer
 from headroom.learn.models import (
-    AnalysisReport,
     ProjectInfo,
     Recommendation,
     RecommendationTarget,
 )
-from headroom.learn.writer import ClaudeCodeWriter, CodexWriter, Recommender
+from headroom.learn.writer import ClaudeCodeWriter, CodexWriter
 
 # =============================================================================
-# Quality Gate Tests (synthetic data, always run)
+# Writer Tests (no LLM needed)
 # =============================================================================
 
 
-class TestQualityGates:
-    """Verify that weak signals don't generate recommendations."""
-
-    def test_single_failure_not_enough(self):
-        """One failure shouldn't generate a recommendation (min_evidence=2)."""
-        recommender = Recommender(min_evidence=2)
-        report = AnalysisReport(
-            project=ProjectInfo(name="test", project_path=Path("/tmp"), data_path=Path("/tmp")),
-            total_calls=100,
-            total_failures=1,
-            total_sessions=1,
-        )
-        # One missing path — below threshold
-        from headroom.learn.models import StructureNote
-
-        report.structure_notes = [
-            StructureNote(
-                category="missing_path",
-                path="/src/foo.py",
-                note="not found",
-                evidence_count=1,
-            )
-        ]
-        recs = recommender.recommend(report)
-        assert recs == []
-
-    def test_low_total_evidence_skipped(self):
-        """Even if individual recs pass, if total evidence is too low, skip all."""
-        recommender = Recommender(min_evidence=1, min_total_evidence=10)
-        report = AnalysisReport(
-            project=ProjectInfo(name="test", project_path=Path("/tmp"), data_path=Path("/tmp")),
-            total_calls=100,
-            total_failures=2,
-            total_sessions=1,
-        )
-        from headroom.learn.models import StructureNote
-
-        report.structure_notes = [
-            StructureNote(category="missing_path", path="/a.py", note="x", evidence_count=2)
-        ]
-        recs = recommender.recommend(report)
-        assert recs == []  # Total evidence=2 < min_total_evidence=10
-
-    def test_sufficient_evidence_passes(self):
-        """Enough evidence generates recommendations."""
-        recommender = Recommender(min_evidence=2, min_total_evidence=3)
-        report = AnalysisReport(
-            project=ProjectInfo(name="test", project_path=Path("/tmp"), data_path=Path("/tmp")),
-            total_calls=100,
-            total_failures=10,
-            total_sessions=5,
-        )
-        from headroom.learn.models import StructureNote
-
-        report.structure_notes = [
-            StructureNote(
-                category="large_file",
-                path="/big.py",
-                note="too big",
-                evidence_count=5,
-                sessions_seen=3,
-            ),
-        ]
-        recs = recommender.recommend(report)
-        assert len(recs) >= 1
-
-    def test_no_failures_no_recommendations(self):
-        """Zero failures = zero recommendations, no files touched."""
-        recommender = Recommender()
-        report = AnalysisReport(
-            project=ProjectInfo(name="test", project_path=Path("/tmp"), data_path=Path("/tmp")),
-            total_calls=500,
-            total_failures=0,
-            total_sessions=10,
-        )
-        recs = recommender.recommend(report)
-        assert recs == []
-
-
 class TestSkipWriteLogic:
     """Verify files are NOT written when there's nothing to write."""
 
@@ -180,10 +102,6 @@ class TestFalsePositiveFiltering:
 
         # Normal code output that happens to contain "error" in identifiers
         assert not is_error_content("def handle_error(e):\n    print('ok')")
-        # Normal sed output with error handling code in the file content
-        assert not is_error_content(
-            '    return fmt.Errorf("connection failed")\n    log.Print("ok")'
-        )
 
     def test_real_error_detected(self):
         """Actual errors should be detected."""
@@ -203,6 +121,7 @@ class TestFalsePositiveFiltering:
 
 CLAUDE_DIR = Path.home() / ".claude" / "projects"
 CODEX_DIR = Path.home() / ".codex" / "sessions"
+HAS_API_KEY = bool(os.environ.get("ANTHROPIC_API_KEY"))
 
 
 @pytest.mark.skipif(not CLAUDE_DIR.exists(), reason="No Claude Code data")
@@ -219,30 +138,37 @@ class TestClaudeCodeIntegration:
             assert p.name
             assert p.data_path.exists()
 
-    def test_full_pipeline_produces_output(self):
-        """Scan → analyze → recommend on real data produces valid output."""
+    def test_scanner_extracts_events(self):
+        """Scanner should extract events including user messages."""
         from headroom.learn.scanner import ClaudeCodeScanner
 
         scanner = ClaudeCodeScanner()
         projects = scanner.discover_projects()
-
-        # Find a project with the most sessions
         best = max(projects, key=lambda p: len(list(p.data_path.glob("*.jsonl"))))
-
         sessions = scanner.scan_project(best)
         assert len(sessions) > 0
 
-        analyzer = FailureAnalyzer()
-        report = analyzer.analyze(best, sessions)
+        # At least some sessions should have events
+        sessions_with_events = [s for s in sessions if s.events]
+        assert len(sessions_with_events) > 0
+
+    @pytest.mark.skipif(not HAS_API_KEY, reason="No ANTHROPIC_API_KEY")
+    def test_full_pipeline_produces_output(self):
+        """Scan → analyze on real data produces valid output."""
+        from headroom.learn.scanner import ClaudeCodeScanner
+
+        scanner = ClaudeCodeScanner()
+        projects = scanner.discover_projects()
+        best = max(projects, key=lambda p: len(list(p.data_path.glob("*.jsonl"))))
+        sessions = scanner.scan_project(best)
+        assert len(sessions) > 0
 
-        assert report.total_calls > 0
-        assert report.total_sessions > 0
-        # Failure rate should be realistic (0-30%)
-        assert 0 <= report.failure_rate <= 0.5
+        analyzer = SessionAnalyzer()
+        result = analyzer.analyze(best, sessions)
 
-        # Corrections should be found if there are failures
-        if report.total_failures > 10:
-            assert len(report.corrections) > 0
+        assert result.total_calls > 0
+        assert result.total_sessions > 0
+        assert 0 <= result.failure_rate <= 0.5
 
     def test_dry_run_writes_nothing(self):
         """Dry run should never create files."""
@@ -252,18 +178,29 @@ class TestClaudeCodeIntegration:
         projects = scanner.discover_projects()
         best = max(projects, key=lambda p: len(list(p.data_path.glob("*.jsonl"))))
 
-        sessions = scanner.scan_project(best)
-        report = FailureAnalyzer().analyze(best, sessions)
-        recs = Recommender().recommend(report)
-
-        writer = ClaudeCodeWriter()
-        result = writer.write(recs, best, dry_run=True)
-
-        assert result.dry_run is True
-        # Verify no new files were created (check a temp path won't exist)
-        for fp in result.files_written:
-            # Files should only be in expected locations
-            assert "CLAUDE.md" in fp.name or "MEMORY.md" in fp.name
+        # Use mock LLM to avoid needing API key
+        mock_response = {
+            "context_file_rules": [
+                {
+                    "section": "Test",
+                    "content": "- test rule",
+                    "estimated_tokens_saved": 100,
+                    "evidence_count": 2,
+                }
+            ],
+            "memory_file_rules": [],
+        }
+        with patch("headroom.learn.analyzer._call_llm", return_value=mock_response):
+            sessions = scanner.scan_project(best)
+            result = SessionAnalyzer().analyze(best, sessions)
+            recs = result.recommendations
+
+            writer = ClaudeCodeWriter()
+            write_result = writer.write(recs, best, dry_run=True)
+
+            assert write_result.dry_run is True
+            for fp in write_result.files_written:
+                assert "CLAUDE.md" in fp.name or "MEMORY.md" in fp.name
 
 
 @pytest.mark.skipif(not CODEX_DIR.exists(), reason="No Codex data")
@@ -287,50 +224,25 @@ class TestCodexIntegration:
 
         assert len(sessions) > 0
 
-        analyzer = FailureAnalyzer()
-        report = analyzer.analyze(projects[0], sessions)
-
-        assert report.total_calls > 0
         # Codex has only Bash tool (shell)
         all_tools = {tc.name for s in sessions for tc in s.tool_calls}
         assert "Bash" in all_tools
 
-        # Failure rate should be realistic
-        assert 0 <= report.failure_rate <= 0.5
-
-    def test_quality_gate_filters_noise(self):
-        """Codex has false positive 'runtime_error' from sed output.
-        Quality gate should prevent these from generating weak recommendations."""
-        from headroom.learn.scanner import CodexScanner
-
-        scanner = CodexScanner()
-        projects = scanner.discover_projects()
-        sessions = scanner.scan_project(projects[0])
-
-        report = FailureAnalyzer().analyze(projects[0], sessions)
-        # Use strict quality gates
-        recommender = Recommender(min_evidence=5, min_total_evidence=10)
-        recs = recommender.recommend(report)
-
-        # Every recommendation should have meaningful evidence
-        for rec in recs:
-            assert rec.evidence_count >= 5
-            assert rec.confidence >= 0.3
-
-    def test_codex_writer_targets_agents_md(self):
+    def test_codex_writer_targets_agents_md(self, tmp_path):
         """Codex writer should target AGENTS.md, not CLAUDE.md."""
-        from headroom.learn.scanner import CodexScanner
-
-        scanner = CodexScanner()
-        projects = scanner.discover_projects()
-        sessions = scanner.scan_project(projects[0])
-
-        report = FailureAnalyzer().analyze(projects[0], sessions)
-        recs = Recommender().recommend(report)
+        proj = ProjectInfo(name="codex-test", project_path=tmp_path, data_path=tmp_path)
+        recs = [
+            Recommendation(
+                target=RecommendationTarget.CONTEXT_FILE,
+                section="Commands",
+                content="- Use npm run test",
+                confidence=0.9,
+                evidence_count=5,
+            )
+        ]
 
-        if recs:
-            writer = CodexWriter()
-            result = writer.write(recs, projects[0], dry_run=True)
-            for fp in result.files_written:
-                assert fp.name in ("AGENTS.md", "instructions.md")
-                assert "CLAUDE.md" not in fp.name
+        writer = CodexWriter()
+        result = writer.write(recs, proj, dry_run=True)
+        for fp in result.files_written:
+            assert fp.name in ("AGENTS.md", "instructions.md")
+            assert "CLAUDE.md" not in fp.name