Add headroom learn: offline failure learning for coding agents

Analyzes past conversation history to find tool call failure patterns,
correlates each failure with what eventually succeeded, and writes
specific project-level learnings to CLAUDE.md and MEMORY.md.

Key design:
- Success correlation: extracts the diff between failed and successful
inputs as the learning (not generic advice)
- Generic architecture: tool-agnostic ToolCall model with pluggable
Scanner/Writer adapters (Claude Code first, extensible to Cursor/Codex)
- 5 analyzers: Environment, Structure, Commands, Retries, Cross-Session
- Dry-run by default, --apply to write, --all for all projects

Also fixes mypy errors in litellm_callback, asgi, langchain chat_model,
and anthropic provider (AsyncClient typing, ToolCall arg-type, int cast).

CHANGELOG.md

@@ -8,6 +8,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **`headroom learn`** — Offline failure learning for coding agents
+  - Analyzes past conversation history (Claude Code, extensible to Cursor/Codex)
+  - **Success correlation**: for each failure, finds what succeeded after and extracts the specific correction
+  - 5 analyzers: Environment, Structure, Command Patterns, Retry Prevention, Cross-Session
+  - Writes specific learnings to CLAUDE.md (stable project facts) and MEMORY.md (session patterns)
+  - Generic architecture: tool-agnostic `ToolCall` model, pluggable Scanner/Writer adapters
+  - Dry-run by default, `--apply` to write, `--all` for all projects
+  - Example output: "FirstClassEntity.java is not at axion-formats/ — actually at axion-scala-common/"
+- **Read Lifecycle Management** — Event-driven compression of stale/superseded Read outputs
+  - Detects when a Read output becomes stale (file was edited after) or superseded (file was re-read)
+  - Replaces stale/superseded content with compact CCR markers, stores originals for retrieval
+  - 75% of Read output bytes are provably stale or redundant (from real-world analysis of 66K tool calls)
+  - Fresh Reads (latest read, no subsequent edit) are never touched — Edit safety preserved
+  - Opt-in via `ReadLifecycleConfig(enabled=True)`, disabled by default
+  - Handles both OpenAI and Anthropic message formats
 - **any-llm backend** - Route requests through 38+ LLM providers (OpenAI, Mistral, Groq, Ollama, etc.) via [any-llm](https://mozilla-ai.github.io/any-llm/providers/)
   - Enable with `--backend anyllm --anyllm-provider <provider>`
   - Install with: `pip install 'headroom-ai[anyllm]'`

README.md

@@ -61,6 +61,16 @@ OPENAI_BASE_URL=http://localhost:8787/v1 cursor
 
 Works with any language, any tool, any framework. One env var. **[Proxy docs](docs/proxy.md)**
 
+### Failure Learning (new)
+
+```bash
+headroom learn                   # Analyze past Claude Code sessions, show recommendations
+headroom learn --apply           # Write learnings to CLAUDE.md and MEMORY.md
+headroom learn --all --apply     # Learn across all your projects
+```
+
+Reads your conversation history, finds every failed tool call, correlates it with what eventually succeeded, and writes specific corrections into your project files. Next session starts smarter. **[Learn docs](docs/learn.md)**
+
 ### Python: One function
 
 ```python
@@ -231,6 +241,8 @@ flowchart TB
 | **Image Compression** | 40-90% token reduction via trained ML router |
 | **Memory** | Persistent memory across conversations |
 | **Compression Hooks** | Customize compression with pre/post hooks |
+| **Read Lifecycle** | Detects stale/superseded Read outputs, replaces with CCR markers |
+| **`headroom learn`** | Analyzes past failures, writes project-specific learnings to CLAUDE.md/MEMORY.md |
 
 ---
 
@@ -276,6 +288,7 @@ Python 3.10+
 | [Memory](docs/memory.md) | Persistent memory |
 | [Agno](docs/agno.md) | Agno agent framework |
 | [MCP](docs/mcp.md) | Claude Code subscriptions |
+| [Learn](docs/learn.md) | Offline failure learning for coding agents |
 | [Configuration](docs/configuration.md) | All options |
 
 ---

docs/learn.md

@@ -0,0 +1,162 @@
+# Headroom Learn
+
+Offline failure learning for coding agents. Analyzes past conversations, finds what went wrong, correlates it with what eventually worked, and writes specific project-level learnings that prevent the same mistakes next session.
+
+## Quick Start
+
+```bash
+# See recommendations for current project (dry-run, no changes)
+headroom learn
+
+# Write recommendations to CLAUDE.md and MEMORY.md
+headroom learn --apply
+
+# Analyze a specific project
+headroom learn --project ~/my-project --apply
+
+# Analyze all projects
+headroom learn --all --apply
+```
+
+## How It Works
+
+```
+Past Sessions → Scanner → Analyzer → Writer → CLAUDE.md / MEMORY.md
+                  │           │          │
+                  │           │          └─ Writes marker-delimited sections
+                  │           │             (replaced on re-run, not duplicated)
+                  │           │
+                  │           └─ Success Correlation: for each failure,
+                  │              finds what succeeded and extracts the diff
+                  │
+                  └─ Reads ~/.claude/projects/*.jsonl
+                     (extensible to Cursor, Codex, etc.)
+```
+
+### Success Correlation
+
+The core innovation. Instead of cataloging failures ("Read failed 5 times"), Headroom finds what the model did to fix each failure:
+
+- **Failed**: `Read axion-formats/src/main/java/.../FirstClassEntity.java`
+- **Then succeeded**: `Read axion-scala-common/src/main/scala/.../FirstClassEntity.scala`
+- **Learning**: "`FirstClassEntity` is at `axion-scala-common/`, not `axion-formats/`"
+
+This produces specific, actionable corrections — not generic advice.
+
+## What It Learns
+
+### 1. Environment Facts → CLAUDE.md
+Which runtime commands work vs fail.
+
+```markdown
+### Environment
+- **Python**: use `uv run python` (not `python3` — modules not available outside venv)
+```
+
+### 2. File Path Corrections → CLAUDE.md
+Wrong paths the model keeps guessing, with the correct locations.
+
+```markdown
+### File Path Corrections
+- `axion-common/src/.../AxionSparkConstants.scala`
+  → actually at `axion-spark-common/src/.../AxionSparkConstants.scala`
+```
+
+### 3. Search Scope → CLAUDE.md
+Which directories to search in (narrow paths fail, broader ones work).
+
+```markdown
+### Search Scope
+- Don't search `axion-model/` → use `axion/` (the repo root)
+```
+
+### 4. Command Patterns → CLAUDE.md
+How commands should (and shouldn't) be run.
+
+```markdown
+### Command Patterns
+- **user_prefers_manual**: User rejected gradle 18 times — show the command, don't execute
+- **python_runtime**: Use `uv run python` not `python3` (ModuleNotFoundError)
+```
+
+### 5. Known Large Files → CLAUDE.md
+Files that need `offset`/`limit` with Read.
+
+```markdown
+### Known Large Files
+- `proxy/server.py` (~8000 lines) — always use offset/limit
+```
+
+### 6. Retry Prevention → MEMORY.md
+Specific suggestions derived from actual corrections.
+
+### 7. Permission Notes → MEMORY.md
+Commands repeatedly rejected — model should suggest them to the user instead.
+
+## Where Learnings Go
+
+| Pattern | Destination | Why |
+|---------|-------------|-----|
+| Environment, paths, search scope, commands, large files | **CLAUDE.md** | Stable project facts, version-controllable |
+| Missing paths, retry patterns, permissions | **MEMORY.md** | May change, agent-specific |
+
+CLAUDE.md lives in your project directory. MEMORY.md lives in `~/.claude/projects/*/memory/`.
+
+## Marker-Based Updates
+
+Headroom manages a clearly-delimited section in each file:
+
+```markdown
+<!-- headroom:learn:start -->
+## Headroom Learned Patterns
+*Auto-generated by `headroom learn` — do not edit manually*
+...
+<!-- headroom:learn:end -->
+```
+
+On re-run, only the content between markers is replaced. Your existing file content is preserved.
+
+## Architecture
+
+```
+Scanner (adapter)  →  Analyzer (generic)  →  Writer (adapter)
+├── ClaudeCodeScanner   ├── EnvironmentAnalyzer   ├── ClaudeCodeWriter
+├── (CursorScanner)     ├── StructureAnalyzer     ├── (CursorWriter)
+└── (GenericScanner)    ├── CommandAnalyzer        └── (GenericWriter)
+                        ├── RetryAnalyzer
+                        └── 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.
+
+To add support for a new agent (e.g., Cursor):
+1. Write `CursorScanner(ConversationScanner)` — reads Cursor's log format
+2. Write `CursorWriter(ContextWriter)` — writes to `.cursorrules`
+3. Same analyzers, same models, same recommendations
+
+## CLI Reference
+
+```
+headroom learn [OPTIONS]
+
+Options:
+  --project PATH    Project directory to analyze (default: current directory)
+  --all             Analyze all discovered projects
+  --apply           Write recommendations (default: dry-run)
+  --claude-dir PATH Path to .claude directory (default: ~/.claude)
+```
+
+## Real-World Results
+
+Tested on 67,583 tool calls across 23 projects:
+
+| Metric | Value |
+|--------|-------|
+| Failure rate | 7.5% (5,066 failures) |
+| Corrections extracted | 164 per project (avg) |
+| Specific path corrections | 22 (axion project) |
+| Search scope corrections | 24 (axion project) |
+| Command patterns learned | 5 (axion project) |
+| Estimated preventable waste | ~27 MB across corpus |

headroom/cli/learn.py

@@ -0,0 +1,148 @@
+"""CLI commands for Headroom Learn — offline failure learning."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import click
+
+from .main import main
+
+
+@main.command()
+@click.option(
+    "--project",
+    type=click.Path(exists=True, path_type=Path),
+    default=None,
+    help="Project directory to analyze. Defaults to current directory.",
+)
+@click.option(
+    "--all",
+    "analyze_all",
+    is_flag=True,
+    default=False,
+    help="Analyze all discovered projects.",
+)
+@click.option(
+    "--apply",
+    is_flag=True,
+    default=False,
+    help="Write recommendations to CLAUDE.md / MEMORY.md (default: dry-run).",
+)
+@click.option(
+    "--claude-dir",
+    type=click.Path(path_type=Path),
+    default=None,
+    help="Path to .claude directory. Defaults to ~/.claude.",
+)
+def learn(
+    project: Path | None,
+    analyze_all: bool,
+    apply: bool,
+    claude_dir: Path | 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.
+
+    \b
+    Examples:
+        headroom learn                    # Analyze current project (dry-run)
+        headroom learn --apply            # Write recommendations
+        headroom learn --all              # Analyze all projects
+        headroom learn --project ~/myapp  # Analyze specific project
+    """
+    from ..learn.analyzer import FailureAnalyzer
+    from ..learn.scanner import ClaudeCodeScanner
+    from ..learn.writer import ClaudeCodeWriter, Recommender
+
+    scanner = ClaudeCodeScanner(claude_dir=claude_dir)
+    analyzer = FailureAnalyzer()
+    recommender = Recommender()
+    writer = ClaudeCodeWriter()
+
+    # Discover projects
+    all_projects = scanner.discover_projects()
+
+    if not all_projects:
+        click.echo("No projects found in ~/.claude/projects/")
+        return
+
+    # Filter to target project(s)
+    if analyze_all:
+        targets = all_projects
+    elif project:
+        resolved = project.resolve()
+        targets = [p for p in all_projects if p.project_path == resolved]
+        if not targets:
+            click.echo(f"Project not found: {resolved}")
+            click.echo(f"Available projects: {', '.join(p.name for p in all_projects)}")
+            return
+    else:
+        # Auto-detect from cwd
+        cwd = Path.cwd().resolve()
+        targets = [p for p in all_projects if p.project_path == cwd]
+        if not targets:
+            # Try parent directories
+            for parent in cwd.parents:
+                targets = [p for p in all_projects if p.project_path == parent]
+                if targets:
+                    break
+        if not targets:
+            click.echo(f"No project data found for {cwd}")
+            click.echo("Try: headroom learn --project <path>  or  headroom learn --all")
+            click.echo("\nAvailable projects:")
+            for p in all_projects[:10]:
+                click.echo(f"  {p.name:30s} {p.project_path}")
+            return
+
+    # Analyze each target
+    for proj in targets:
+        click.echo(f"\n{'=' * 60}")
+        click.echo(f"Project: {proj.name}")
+        click.echo(f"Path:    {proj.project_path}")
+        click.echo(f"{'=' * 60}")
+
+        sessions = scanner.scan_project(proj)
+        if not sessions:
+            click.echo("  No conversation data found.")
+            continue
+
+        report = analyzer.analyze(proj, sessions)
+
+        # Print summary
+        click.echo(f"\n  Sessions analyzed: {report.total_sessions}")
+        click.echo(f"  Total tool calls:  {report.total_calls}")
+        click.echo(f"  Failed calls:      {report.total_failures} ({report.failure_rate:.1%})")
+        click.echo(f"  Waste bytes:       {report.waste_bytes / 1024:.0f} KB")
+
+        if report.failure_rate == 0:
+            click.echo("\n  No failures found. Nothing to learn.")
+            continue
+
+        # Generate recommendations
+        recommendations = recommender.recommend(report)
+
+        if not recommendations:
+            click.echo("\n  No actionable patterns found.")
+            continue
+
+        click.echo(f"\n  Recommendations: {len(recommendations)}")
+
+        # Write (or dry-run)
+        result = writer.write(recommendations, proj, dry_run=not apply)
+
+        for file_path, content in result.content_by_file.items():
+            click.echo(f"\n  {'[WOULD WRITE]' if result.dry_run else '[WROTE]'} {file_path}")
+            click.echo(f"  {'─' * 50}")
+            # Show content preview (indented)
+            for line in content.split("\n"):
+                if line.startswith("<!-- headroom"):
+                    continue  # Skip markers in display
+                click.echo(f"  {line}")
+            click.echo(f"  {'─' * 50}")
+
+        if result.dry_run:
+            click.echo("\n  Dry run — no files modified. Use --apply to write.")

headroom/cli/main.py

@@ -35,6 +35,7 @@ def _register_commands() -> None:
     """Register all subcommand groups."""
     from . import (
         evals,  # noqa: F401
+        learn,  # noqa: F401
         mcp,  # noqa: F401
         memory,  # noqa: F401
         proxy,  # noqa: F401

headroom/integrations/asgi.py

@@ -80,7 +80,7 @@ class CompressionMiddleware:
         self._api_url = (
             api_url or os.environ.get("HEADROOM_API_URL", "").strip() or _DEFAULT_CLOUD_URL
         ).rstrip("/")
-        self._client = None  # Lazy-initialized httpx.AsyncClient
+        self._client: Any = None  # Lazy-initialized httpx.AsyncClient
 
     @property
     def cloud_mode(self) -> bool:

headroom/integrations/langchain/chat_model.py

@@ -79,7 +79,7 @@ def _check_langchain_available() -> None:
         )
 
 
-def _tool_call_args_to_json(tc: dict[str, Any]) -> str:
+def _tool_call_args_to_json(tc: dict[str, Any] | Any) -> str:
     """Normalize tool call arguments to JSON string for OpenAI format.
 
     LangChain can provide 'args' (dict) or 'arguments' (str) depending on source.

headroom/integrations/litellm_callback.py

@@ -69,7 +69,7 @@ class HeadroomCallback:
         self._api_url = (
             api_url or os.environ.get("HEADROOM_API_URL", "").strip() or _DEFAULT_CLOUD_URL
         ).rstrip("/")
-        self._client = None  # Lazy-initialized httpx.AsyncClient
+        self._client: Any = None  # Lazy-initialized httpx.AsyncClient
 
     @property
     def total_tokens_saved(self) -> int:

headroom/learn/__init__.py

@@ -0,0 +1,17 @@
+"""Headroom Learn — offline failure 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.
+
+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.
+"""

headroom/learn/analyzer.py

@@ -0,0 +1,618 @@
+"""Failure analyzers with success correlation.
+
+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.
+
+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.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from collections import Counter, defaultdict
+
+from .models import (
+    AnalysisReport,
+    CommandPattern,
+    Correction,
+    EnvironmentFact,
+    ErrorCategory,
+    ProjectInfo,
+    RetryPattern,
+    SessionData,
+    StructureNote,
+    ToolCall,
+)
+
+# How many messages ahead to look for a success after a failure
+_CORRECTION_WINDOW = 10
+
+
+class FailureAnalyzer:
+    """Runs all analyzers on tool call data and produces an AnalysisReport."""
+
+    def analyze(self, project: ProjectInfo, sessions: list[SessionData]) -> AnalysisReport:
+        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(
+            project=project,
+            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
+
+
+# =============================================================================
+# Success Correlation: The Core Learning Primitive
+# =============================================================================
+
+
+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
+
+
+# =============================================================================
+# Environment Analyzer (uses corrections for python/build tool detection)
+# =============================================================================
+
+
+def _analyze_environment(sessions: list[SessionData]) -> list[EnvironmentFact]:
+    """Detect which runtime commands work vs fail."""
+    facts: list[EnvironmentFact] = []
+
+    python_failures: Counter[str] = Counter()
+    python_successes: Counter[str] = Counter()
+    python_sessions: dict[str, set[str]] = defaultdict(set)
+
+    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,
+                )
+            )
+
+    return facts
+
+
+# =============================================================================
+# Structure Analyzer (uses corrections to learn correct paths)
+# =============================================================================
+
+
+def _analyze_structure(
+    sessions: list[SessionData], corrections: list[Correction]
+) -> list[StructureNote]:
+    """Find file structure issues and learn correct paths from corrections."""
+    notes: list[StructureNote] = []
+
+    # 1. Path corrections: wrong path → correct path (from success correlation)
+    path_corrections: dict[str, Counter[str]] = defaultdict(Counter)
+    for c in corrections:
+        if c.tool_name not in ("Read", "read"):
+            continue
+        if c.error_category != ErrorCategory.FILE_NOT_FOUND:
+            continue
+        failed_path = c.failed_input.get("file_path", "")
+        success_path = c.success_input.get("file_path", "")
+        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,
+            )
+        )
+
+    # 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,
+            )
+        )
+
+    # 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]),
+            )
+        )
+
+    # 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]),
+            )
+        )
+
+    return notes
+
+
+# =============================================================================
+# 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
+                        }
+                    ),
+                )
+            )
+
+    # 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,
+                )
+            )
+
+    # 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
+
+
+# =============================================================================
+# Retry Analyzer (uses corrections to provide specific suggestions)
+# =============================================================================
+
+
+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,
+            )
+        )
+
+    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")
+
+
+# =============================================================================
+# Permission Analyzer
+# =============================================================================
+
+
+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
+
+    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
+            )
+            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
+# =============================================================================
+
+
+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)
+
+    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
+
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+
+_PYTHON_CMD_RE = re.compile(
+    r"^((?:source\s+\S+\s*&&\s*)?(?:\.venv/bin/)?(?:python3?|uv run python|uv run|/opt/nflx/python))"
+)
+
+
+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 _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

headroom/learn/models.py

@@ -0,0 +1,240 @@
+"""Data models for Headroom Learn — tool-agnostic abstractions.
+
+These models normalize tool call data from ANY agent system (Claude Code, Cursor,
+Codex, custom agents) into a common format that analyzers can work with.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from pathlib import Path
+
+# =============================================================================
+# Error Classification
+# =============================================================================
+
+
+class ErrorCategory(str, Enum):
+    """Classified error categories for tool call failures."""
+
+    FILE_NOT_FOUND = "file_not_found"
+    MODULE_NOT_FOUND = "module_not_found"
+    COMMAND_NOT_FOUND = "command_not_found"
+    PERMISSION_DENIED = "permission_denied"
+    FILE_TOO_LARGE = "file_too_large"
+    IS_DIRECTORY = "is_directory"
+    SYNTAX_ERROR = "syntax_error"
+    RUNTIME_ERROR = "runtime_error"
+    TIMEOUT = "timeout"
+    NO_MATCHES = "no_matches"  # Grep/Glob found nothing
+    USER_REJECTED = "user_rejected"
+    SIBLING_ERROR = "sibling_error"  # Cascade from parallel call failure
+    EXIT_CODE = "exit_code"
+    CONNECTION_ERROR = "connection_error"
+    BUILD_FAILURE = "build_failure"
+    UNKNOWN = "unknown"
+
+
+# =============================================================================
+# Core Data Models (Tool-Agnostic)
+# =============================================================================
+
+
+@dataclass
+class ToolCall:
+    """A single tool call and its result — normalized from any agent system.
+
+    This is the fundamental unit of analysis. Scanners produce these,
+    analyzers consume them.
+    """
+
+    name: str  # Tool name ("Bash", "Read", "file_search", etc.)
+    tool_call_id: str  # Unique ID linking call to result
+    input_data: dict  # Tool input parameters
+    output: str  # Result content (may be error message)
+    is_error: bool  # Whether the call failed
+    error_category: ErrorCategory = ErrorCategory.UNKNOWN
+    msg_index: int = 0  # Position in conversation
+    output_bytes: int = 0  # Size of output
+
+    @property
+    def input_summary(self) -> str:
+        """Short summary of tool input for display."""
+        if self.name in ("Bash", "bash"):
+            cmd: str = self.input_data.get("command", "")
+            return cmd[:100] + "..." if len(cmd) > 100 else cmd
+        if self.name in ("Read", "read"):
+            return str(self.input_data.get("file_path", "?"))
+        if self.name in ("Grep", "grep"):
+            return str(self.input_data.get("pattern", "?"))
+        if self.name in ("Glob", "glob"):
+            return str(self.input_data.get("pattern", "?"))
+        if self.name in ("Edit", "edit", "Write", "write"):
+            return str(self.input_data.get("file_path", "?"))
+        return str(self.input_data)[:80]
+
+
+@dataclass
+class SessionData:
+    """Normalized data from a single conversation session."""
+
+    session_id: str
+    tool_calls: list[ToolCall] = field(default_factory=list)
+    timestamp: datetime | None = None
+
+    @property
+    def failure_count(self) -> int:
+        return sum(1 for tc in self.tool_calls if tc.is_error)
+
+    @property
+    def failure_rate(self) -> float:
+        if not self.tool_calls:
+            return 0.0
+        return self.failure_count / len(self.tool_calls)
+
+
+@dataclass
+class ProjectInfo:
+    """Information about a project discovered by a scanner."""
+
+    name: str  # Human-readable project name
+    project_path: Path  # Actual project directory
+    data_path: Path  # Where conversation logs are stored
+    context_file: Path | None = None  # CLAUDE.md / .cursorrules / AGENTS.md
+    memory_file: Path | None = None  # MEMORY.md or equivalent
+
+
+# =============================================================================
+# Analysis Output Models
+# =============================================================================
+
+
+class RecommendationTarget(str, Enum):
+    """Where a recommendation should be written."""
+
+    CONTEXT_FILE = "context_file"  # CLAUDE.md, .cursorrules, AGENTS.md
+    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."""
+
+    target: RecommendationTarget
+    section: str  # Section heading (e.g., "Environment", "Known Large Files")
+    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
+
+
+@dataclass
+class AnalysisReport:
+    """Complete output of failure analysis for a project."""
+
+    project: ProjectInfo
+    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)
+
+    @property
+    def failure_rate(self) -> float:
+        if not self.total_calls:
+            return 0.0
+        return self.total_failures / self.total_calls

headroom/learn/scanner.py

@@ -0,0 +1,358 @@
+"""Conversation scanners — read tool call logs from different agent systems.
+
+Scanners normalize conversation data into ToolCall sequences that analyzers
+can process regardless of the source system.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+from .models import (
+    ErrorCategory,
+    ProjectInfo,
+    SessionData,
+    ToolCall,
+)
+
+logger = logging.getLogger(__name__)
+
+# =============================================================================
+# Error Classification
+# =============================================================================
+
+# Patterns checked in order — first match wins
+_ERROR_PATTERNS: list[tuple[re.Pattern, ErrorCategory]] = [
+    (
+        re.compile(r"No such file or directory|ENOENT|FileNotFoundError|does not exist", re.I),
+        ErrorCategory.FILE_NOT_FOUND,
+    ),
+    (
+        re.compile(r"ModuleNotFoundError|ImportError|No module named", re.I),
+        ErrorCategory.MODULE_NOT_FOUND,
+    ),
+    (re.compile(r"command not found", re.I), ErrorCategory.COMMAND_NOT_FOUND),
+    (
+        re.compile(r"Permission denied|EACCES|EPERM|auto-denied", re.I),
+        ErrorCategory.PERMISSION_DENIED,
+    ),
+    (
+        re.compile(r"file is too large|too many lines|exceeds.*limit", re.I),
+        ErrorCategory.FILE_TOO_LARGE,
+    ),
+    (re.compile(r"EISDIR|Is a directory", re.I), ErrorCategory.IS_DIRECTORY),
+    (re.compile(r"SyntaxError|IndentationError", re.I), ErrorCategory.SYNTAX_ERROR),
+    (re.compile(r"Traceback \(most recent|Exception:|Error:", re.I), ErrorCategory.RUNTIME_ERROR),
+    (re.compile(r"timed? ?out|TimeoutError|deadline exceeded", re.I), ErrorCategory.TIMEOUT),
+    (re.compile(r"No (?:matches|files|results) found|0 matches", re.I), ErrorCategory.NO_MATCHES),
+    (
+        re.compile(r"user.*reject|user.*denied|declined|didn't want to proceed", re.I),
+        ErrorCategory.USER_REJECTED,
+    ),
+    (re.compile(r"[Ss]ibling tool call errored", re.I), ErrorCategory.SIBLING_ERROR),
+    (re.compile(r"exit code|non-zero|exited with", re.I), ErrorCategory.EXIT_CODE),
+    (
+        re.compile(r"ConnectionError|ConnectionRefused|ECONNREFUSED|network", re.I),
+        ErrorCategory.CONNECTION_ERROR,
+    ),
+    (
+        re.compile(r"BUILD FAILED|compilation error|compile error", re.I),
+        ErrorCategory.BUILD_FAILURE,
+    ),
+]
+
+
+def classify_error(content: str) -> ErrorCategory:
+    """Classify an error message into a category."""
+    for pattern, category in _ERROR_PATTERNS:
+        if pattern.search(content[:2000]):  # Only check first 2KB
+            return category
+    return ErrorCategory.UNKNOWN
+
+
+def is_error_content(content: str) -> bool:
+    """Heuristic: does this tool result look like an error?"""
+    if not content or len(content) < 10:
+        return False
+    # Check for common error indicators in first 1KB
+    snippet = content[:1000]
+    indicators = [
+        "Error:",
+        "error:",
+        "ENOENT",
+        "No such file",
+        "command not found",
+        "Permission denied",
+        "ModuleNotFoundError",
+        "Traceback (most recent",
+        "FAILED",
+        "EISDIR",
+        "auto-denied",
+        "Sibling tool call errored",
+        "timed out",
+        "exit code",
+        "FileNotFoundError",
+    ]
+    return any(ind in snippet for ind in indicators)
+
+
+# =============================================================================
+# Abstract Scanner
+# =============================================================================
+
+
+class ConversationScanner(ABC):
+    """Base class for scanning conversation logs from any agent system.
+
+    Subclasses implement log format parsing for specific tools (Claude Code,
+    Cursor, Codex, etc.) and produce normalized ToolCall sequences.
+    """
+
+    @abstractmethod
+    def discover_projects(self) -> list[ProjectInfo]:
+        """Discover all projects with conversation data."""
+        ...
+
+    @abstractmethod
+    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
+        """Scan all sessions for a project, returning normalized tool calls."""
+        ...
+
+
+# =============================================================================
+# Claude Code Scanner
+# =============================================================================
+
+
+class ClaudeCodeScanner(ConversationScanner):
+    """Reads Claude Code conversation logs from ~/.claude/projects/.
+
+    Claude Code stores conversations as JSONL files with these line types:
+    - type="assistant": message.content[] has tool_use blocks (name, input, id)
+    - type="user": message.content[] has tool_result blocks (tool_use_id, content)
+    """
+
+    def __init__(self, claude_dir: Path | None = None):
+        self.claude_dir = claude_dir or Path.home() / ".claude"
+        self.projects_dir = self.claude_dir / "projects"
+
+    def discover_projects(self) -> list[ProjectInfo]:
+        """Discover all projects under ~/.claude/projects/."""
+        if not self.projects_dir.exists():
+            return []
+
+        projects = []
+        for entry in sorted(self.projects_dir.iterdir()):
+            if not entry.is_dir() or entry.name.startswith("."):
+                continue
+
+            # Decode project path from escaped directory name
+            # e.g., "-Users-tchopra-claude-projects-headroom" → "/Users/tchopra/claude-projects/headroom"
+            project_path = Path("/" + entry.name.replace("-", "/", entry.name.count("-")))
+
+            # Try smarter decoding: split on segments that look like path components
+            # The escaping replaces / with - but also - in names stays as -
+            # Heuristic: try the decoded path, if it exists use it
+            decoded = _decode_project_path(entry.name)
+            if decoded:
+                project_path = decoded
+
+            # Derive human-readable name
+            name = project_path.name if project_path != Path("/") else entry.name
+
+            # Check for CLAUDE.md in actual project directory
+            context_file = None
+            if project_path.exists():
+                claude_md = project_path / "CLAUDE.md"
+                if claude_md.exists():
+                    context_file = claude_md
+
+            # Check for MEMORY.md
+            memory_dir = entry / "memory"
+            memory_file = memory_dir / "MEMORY.md" if memory_dir.exists() else None
+            if memory_file and not memory_file.exists():
+                memory_file = None
+
+            # Only include projects with JSONL files
+            jsonl_files = list(entry.glob("*.jsonl"))
+            if not jsonl_files:
+                continue
+
+            projects.append(
+                ProjectInfo(
+                    name=name,
+                    project_path=project_path,
+                    data_path=entry,
+                    context_file=context_file,
+                    memory_file=memory_file,
+                )
+            )
+
+        return projects
+
+    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
+        """Scan all conversation JSONL files for a project."""
+        sessions = []
+
+        # Find all JSONL files (main conversations, not subagent files)
+        jsonl_files = sorted(project.data_path.glob("*.jsonl"))
+
+        for jsonl_path in jsonl_files:
+            session = self._scan_session(jsonl_path)
+            if session and session.tool_calls:
+                sessions.append(session)
+
+        return sessions
+
+    def _scan_session(self, jsonl_path: Path) -> SessionData | None:
+        """Scan a single JSONL conversation file."""
+        session_id = jsonl_path.stem
+        tool_uses: dict[str, tuple[str, dict]] = {}  # tc_id → (tool_name, input)
+        tool_calls: list[ToolCall] = []
+        msg_index = 0
+
+        try:
+            with open(jsonl_path) as f:
+                for line in f:
+                    try:
+                        d = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+
+                    msg_index += 1
+                    line_type = d.get("type", "")
+
+                    if line_type == "assistant":
+                        self._extract_tool_uses(d, tool_uses)
+                    elif line_type == "user":
+                        self._extract_tool_results(d, tool_uses, tool_calls, msg_index)
+
+        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)
+
+    def _extract_tool_uses(self, d: dict, tool_uses: dict[str, tuple[str, dict]]) -> None:
+        """Extract tool_use blocks from an assistant message."""
+        msg = d.get("message", {})
+        content = msg.get("content", [])
+        if not isinstance(content, list):
+            return
+
+        for block in content:
+            if not isinstance(block, dict) or block.get("type") != "tool_use":
+                continue
+            tc_id = block.get("id", "")
+            name = block.get("name", "")
+            inp = block.get("input", {})
+            if tc_id and name:
+                tool_uses[tc_id] = (name, inp if isinstance(inp, dict) else {})
+
+    def _extract_tool_results(
+        self,
+        d: dict,
+        tool_uses: dict[str, tuple[str, dict]],
+        tool_calls: list[ToolCall],
+        msg_index: int,
+    ) -> None:
+        """Extract tool_result blocks from a user message and match to tool_uses."""
+        msg = d.get("message", {})
+        content = msg.get("content", [])
+        if not isinstance(content, list):
+            return
+
+        for block in content:
+            if not isinstance(block, dict) or block.get("type") != "tool_result":
+                continue
+
+            tc_id = block.get("tool_use_id", "")
+            result_content = block.get("content", "")
+            if not isinstance(result_content, str):
+                result_content = str(result_content)
+
+            # Match to tool_use
+            if tc_id not in tool_uses:
+                continue
+
+            name, inp = tool_uses[tc_id]
+
+            # Determine if error
+            explicit_error = block.get("is_error", False)
+            detected_error = is_error_content(result_content)
+            is_err = explicit_error or detected_error
+
+            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,
+                    msg_index=msg_index,
+                    output_bytes=len(result_content.encode("utf-8")),
+                )
+            )
+
+
+def _decode_project_path(escaped_name: str) -> Path | None:
+    """Decode a Claude Code escaped project path.
+
+    Claude Code escapes paths by replacing / with -.
+    e.g., "-Users-tchopra-claude-projects-headroom"
+    → "/Users/tchopra/claude-projects/headroom"
+
+    Since - is ambiguous (path separator vs literal hyphen), we try
+    progressively and check which decoded path actually exists.
+    """
+    if not escaped_name.startswith("-"):
+        return None
+
+    # Simple approach: replace all - with / and check if path exists
+    simple = Path("/" + escaped_name[1:].replace("-", "/"))
+    if simple.exists():
+        return simple
+
+    # Try common patterns: /Users/username/...
+    parts = escaped_name[1:].split("-")
+    if len(parts) < 3:
+        return None
+
+    # Build path greedily: try joining with / and check existence
+    # Start with /Users/username (first 2 components are almost always correct)
+    if parts[0] == "Users" and len(parts) > 2:
+        base = Path(f"/{parts[0]}/{parts[1]}")
+        remaining = parts[2:]
+        return _greedy_path_decode(base, remaining)
+
+    return None
+
+
+def _greedy_path_decode(base: Path, parts: list[str]) -> Path | None:
+    """Greedily decode remaining path parts, trying - as / first."""
+    if not parts:
+        return base if base.exists() else None
+
+    # Try using / (this part is a directory component)
+    slash_path = base / parts[0]
+    result = _greedy_path_decode(slash_path, parts[1:])
+    if result:
+        return result
+
+    # Try joining with - (this part has a literal hyphen)
+    if len(parts) > 1:
+        hyphen_name = f"{parts[0]}-{parts[1]}"
+        hyphen_path = base / hyphen_name
+        result = _greedy_path_decode(hyphen_path, parts[2:])
+        if result:
+            return result
+
+    # If we've exhausted parts, check if current path exists
+    return base if base.exists() else None

headroom/learn/writer.py

@@ -0,0 +1,333 @@
+"""Context writers — write learned patterns to agent-specific context files.
+
+Writers take Recommendations and write them to the appropriate context
+injection mechanism for each agent system (CLAUDE.md, .cursorrules, etc.).
+"""
+
+from __future__ import annotations
+
+import re
+from abc import ABC, abstractmethod
+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
+_MARKER_START = "<!-- headroom:learn:start -->"
+_MARKER_END = "<!-- headroom:learn:end -->"
+_MARKER_PATTERN = re.compile(
+    re.escape(_MARKER_START) + r".*?" + re.escape(_MARKER_END),
+    re.DOTALL,
+)
+
+
+# =============================================================================
+# Recommender: AnalysisReport → Recommendations
+# =============================================================================
+
+
+class Recommender:
+    """Converts an AnalysisReport into concrete markdown recommendations."""
+
+    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),
+                )
+            )
+
+        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
+# =============================================================================
+
+
+class ContextWriter(ABC):
+    """Base class for writing recommendations to context/memory files."""
+
+    @abstractmethod
+    def write(
+        self,
+        recommendations: list[Recommendation],
+        project: ProjectInfo,
+        dry_run: bool = True,
+    ) -> WriteResult: ...
+
+
+# =============================================================================
+# Write Result
+# =============================================================================
+
+
+class WriteResult:
+    """Result of a write operation."""
+
+    def __init__(self) -> None:
+        self.files_written: list[Path] = []
+        self.content_by_file: dict[Path, str] = {}
+        self.dry_run: bool = True
+
+    def add(self, path: Path, content: str) -> None:
+        self.files_written.append(path)
+        self.content_by_file[path] = content
+
+
+# =============================================================================
+# Claude Code Writer
+# =============================================================================
+
+
+class ClaudeCodeWriter(ContextWriter):
+    """Writes learned patterns to CLAUDE.md and MEMORY.md for Claude Code."""
+
+    def write(
+        self,
+        recommendations: list[Recommendation],
+        project: ProjectInfo,
+        dry_run: bool = True,
+    ) -> WriteResult:
+        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)
+            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)
+            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)
+
+        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"

headroom/providers/anthropic.py

@@ -487,11 +487,11 @@ class AnthropicProvider(Provider):
                 info = litellm_get_model_info(model)
                 if info:
                     if "max_input_tokens" in info and info["max_input_tokens"] is not None:
-                        limit = info["max_input_tokens"]
+                        limit = int(info["max_input_tokens"])
                         self._context_limits[model] = limit
                         return limit
                     if "max_tokens" in info and info["max_tokens"] is not None:
-                        limit = info["max_tokens"]
+                        limit = int(info["max_tokens"])
                         self._context_limits[model] = limit
                         return limit
             except Exception as e:

tests/test_learn/test_analyzer.py

@@ -0,0 +1,348 @@
+"""Tests for failure analyzers — generic tool call pattern recognition."""
+
+from pathlib import Path
+
+from headroom.learn.analyzer import FailureAnalyzer
+from headroom.learn.models import (
+    ErrorCategory,
+    ProjectInfo,
+    SessionData,
+    ToolCall,
+)
+
+
+def _project() -> ProjectInfo:
+    return ProjectInfo(
+        name="test-project",
+        project_path=Path("/tmp/test-project"),
+        data_path=Path("/tmp/test-data"),
+    )
+
+
+def _tc(
+    name: str = "Bash",
+    input_data: dict | None = None,
+    output: str = "ok",
+    is_error: bool = False,
+    error_category: ErrorCategory = ErrorCategory.UNKNOWN,
+    msg_index: int = 0,
+) -> ToolCall:
+    return ToolCall(
+        name=name,
+        tool_call_id=f"tc_{msg_index}",
+        input_data=input_data or {},
+        output=output,
+        is_error=is_error,
+        error_category=error_category,
+        msg_index=msg_index,
+        output_bytes=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
+
+    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()
+        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),
+                ],
+            )
+        ]
+        report = analyzer.analyze(_project(), sessions)
+        assert report.total_calls == 3
+        assert report.total_failures == 1
+
+
+class TestEnvironmentAnalyzer:
+    def test_detects_wrong_python(self):
+        """Module not found with python3 + successes with uv run → learn correct command."""
+        analyzer = FailureAnalyzer()
+        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,
+                        msg_index=0,
+                    ),
+                    _tc(
+                        name="Bash",
+                        input_data={"command": "python3 -c 'import mylib'"},
+                        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
+
+
+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,
+                    ),
+                ],
+            ),
+        ]
+        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
+
+    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,
+                    ),
+                ],
+            ),
+        ]
+        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
+
+    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
+
+
+class TestRetryAnalyzer:
+    def test_detects_stubborn_retries(self):
+        """Same tool failing 3+ times in a row → retry pattern."""
+        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(5)
+                ],
+            )
+        ]
+        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()
+        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
+                ],
+            )
+        ]
+        report = analyzer.analyze(_project(), sessions)
+        assert len(report.retry_patterns) == 0
+
+
+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)
+
+    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)
+        ]
+        report = analyzer.analyze(_project(), sessions)
+        assert len(report.cross_session_patterns) == 0
+
+
+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

tests/test_learn/test_writer.py

@@ -0,0 +1,115 @@
+"""Tests for recommendation writer — marker-based file updates."""
+
+from pathlib import Path
+
+from headroom.learn.models import ProjectInfo, Recommendation, RecommendationTarget
+from headroom.learn.writer import _MARKER_END, _MARKER_START, ClaudeCodeWriter
+
+
+def _project(tmp_path: Path) -> ProjectInfo:
+    proj_dir = tmp_path / "myproject"
+    proj_dir.mkdir()
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    memory_dir = data_dir / "memory"
+    memory_dir.mkdir()
+    return ProjectInfo(
+        name="myproject",
+        project_path=proj_dir,
+        data_path=data_dir,
+    )
+
+
+def _rec(target: RecommendationTarget, section: str, content: str) -> Recommendation:
+    return Recommendation(
+        target=target, section=section, content=content, confidence=0.8, evidence_count=5
+    )
+
+
+class TestClaudeCodeWriter:
+    def test_dry_run_does_not_write(self, tmp_path):
+        proj = _project(tmp_path)
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.CONTEXT_FILE, "Environment", "- Use uv")]
+
+        result = writer.write(recs, proj, dry_run=True)
+
+        assert result.dry_run is True
+        assert len(result.files_written) == 1
+        # File should NOT exist (dry run)
+        claude_md = proj.project_path / "CLAUDE.md"
+        assert not claude_md.exists()
+
+    def test_apply_writes_claude_md(self, tmp_path):
+        proj = _project(tmp_path)
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.CONTEXT_FILE, "Environment", "- Use `uv run python`")]
+
+        result = writer.write(recs, proj, dry_run=False)
+
+        assert result.dry_run is False
+        claude_md = proj.project_path / "CLAUDE.md"
+        assert claude_md.exists()
+        content = claude_md.read_text()
+        assert "uv run python" in content
+        assert _MARKER_START in content
+        assert _MARKER_END in content
+
+    def test_apply_writes_memory_md(self, tmp_path):
+        proj = _project(tmp_path)
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.MEMORY_FILE, "Retry Prevention", "- Don't retry globs")]
+
+        writer.write(recs, proj, dry_run=False)
+
+        memory_md = proj.data_path / "memory" / "MEMORY.md"
+        assert memory_md.exists()
+        assert "Don't retry globs" in memory_md.read_text()
+
+    def test_preserves_existing_claude_md_content(self, tmp_path):
+        proj = _project(tmp_path)
+        claude_md = proj.project_path / "CLAUDE.md"
+        claude_md.write_text("# My Project\n\nExisting instructions here.\n")
+
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.CONTEXT_FILE, "Environment", "- Use uv")]
+        writer.write(recs, proj, dry_run=False)
+
+        content = claude_md.read_text()
+        assert "My Project" in content
+        assert "Existing instructions here" in content
+        assert "Use uv" in content
+
+    def test_replaces_existing_headroom_section(self, tmp_path):
+        proj = _project(tmp_path)
+        claude_md = proj.project_path / "CLAUDE.md"
+        old_section = (
+            f"# My Project\n\n{_MARKER_START}\n## Old Patterns\nold stuff\n{_MARKER_END}\n"
+        )
+        claude_md.write_text(old_section)
+
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.CONTEXT_FILE, "Environment", "- New stuff")]
+        writer.write(recs, proj, dry_run=False)
+
+        content = claude_md.read_text()
+        assert "old stuff" not in content
+        assert "New stuff" in content
+        assert "My Project" in content
+        # Should have exactly one marker pair
+        assert content.count(_MARKER_START) == 1
+        assert content.count(_MARKER_END) == 1
+
+    def test_appends_to_existing_memory_md(self, tmp_path):
+        proj = _project(tmp_path)
+        memory_md = proj.data_path / "memory" / "MEMORY.md"
+        memory_md.write_text("# Existing Memory\n\nSome facts.\n")
+
+        writer = ClaudeCodeWriter()
+        recs = [_rec(RecommendationTarget.MEMORY_FILE, "Retry Prevention", "- New pattern")]
+        writer.write(recs, proj, dry_run=False)
+
+        content = memory_md.read_text()
+        assert "Existing Memory" in content
+        assert "Some facts" in content
+        assert "New pattern" in content