From 17442c2dcc5ba532fff9ac1aa715fbb677a18b9c Mon Sep 17 00:00:00 2001 From: chopratejas Date: Fri, 27 Feb 2026 21:18:33 -0800 Subject: [PATCH] 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 | 15 + README.md | 13 + docs/learn.md | 162 +++++ headroom/cli/learn.py | 148 +++++ headroom/cli/main.py | 1 + headroom/integrations/asgi.py | 2 +- headroom/integrations/langchain/chat_model.py | 2 +- headroom/integrations/litellm_callback.py | 2 +- headroom/learn/__init__.py | 17 + headroom/learn/analyzer.py | 618 ++++++++++++++++++ headroom/learn/models.py | 240 +++++++ headroom/learn/scanner.py | 358 ++++++++++ headroom/learn/writer.py | 333 ++++++++++ headroom/providers/anthropic.py | 4 +- tests/test_learn/__init__.py | 0 tests/test_learn/test_analyzer.py | 348 ++++++++++ tests/test_learn/test_writer.py | 115 ++++ 17 files changed, 2373 insertions(+), 5 deletions(-) create mode 100644 docs/learn.md create mode 100644 headroom/cli/learn.py create mode 100644 headroom/learn/__init__.py create mode 100644 headroom/learn/analyzer.py create mode 100644 headroom/learn/models.py create mode 100644 headroom/learn/scanner.py create mode 100644 headroom/learn/writer.py create mode 100644 tests/test_learn/__init__.py create mode 100644 tests/test_learn/test_analyzer.py create mode 100644 tests/test_learn/test_writer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d66d70929..1a68f4ac2 100644 --- a/CHANGELOG.md +++ b/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 ` - Install with: `pip install 'headroom-ai[anyllm]'` diff --git a/README.md b/README.md index 8c082d5d0..cd933f838 100644 --- a/README.md +++ b/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 | --- diff --git a/docs/learn.md b/docs/learn.md new file mode 100644 index 000000000..dce072a81 --- /dev/null +++ b/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 Learned Patterns +*Auto-generated by `headroom learn` — do not edit manually* +... + +``` + +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 | diff --git a/headroom/cli/learn.py b/headroom/cli/learn.py new file mode 100644 index 000000000..8671f9fbc --- /dev/null +++ b/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 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("" +_MARKER_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" diff --git a/headroom/providers/anthropic.py b/headroom/providers/anthropic.py index 85df9961d..e65fa854c 100644 --- a/headroom/providers/anthropic.py +++ b/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: diff --git a/tests/test_learn/__init__.py b/tests/test_learn/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_learn/test_analyzer.py b/tests/test_learn/test_analyzer.py new file mode 100644 index 000000000..fcdd8acdc --- /dev/null +++ b/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 diff --git a/tests/test_learn/test_writer.py b/tests/test_learn/test_writer.py new file mode 100644 index 000000000..6bb12b832 --- /dev/null +++ b/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