mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
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).
This commit is contained in:
chopratejas
parent
40369762dd
commit
17442c2dcc
17 changed files with 2373 additions and 5 deletions
15
CHANGELOG.md
15
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]'`
|
||||
|
|
|
|||
13
README.md
13
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 |
|
||||
|
||||
---
|
||||
|
|
|
|||
162
docs/learn.md
Normal file
162
docs/learn.md
Normal file
|
|
@ -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 |
|
||||
148
headroom/cli/learn.py
Normal file
148
headroom/cli/learn.py
Normal file
|
|
@ -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.")
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
17
headroom/learn/__init__.py
Normal file
17
headroom/learn/__init__.py
Normal file
|
|
@ -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.
|
||||
"""
|
||||
618
headroom/learn/analyzer.py
Normal file
618
headroom/learn/analyzer.py
Normal file
|
|
@ -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
|
||||
240
headroom/learn/models.py
Normal file
240
headroom/learn/models.py
Normal file
|
|
@ -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
|
||||
358
headroom/learn/scanner.py
Normal file
358
headroom/learn/scanner.py
Normal file
|
|
@ -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
|
||||
333
headroom/learn/writer.py
Normal file
333
headroom/learn/writer.py
Normal file
|
|
@ -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"
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
0
tests/test_learn/__init__.py
Normal file
0
tests/test_learn/__init__.py
Normal file
348
tests/test_learn/test_analyzer.py
Normal file
348
tests/test_learn/test_analyzer.py
Normal file
|
|
@ -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
|
||||
115
tests/test_learn/test_writer.py
Normal file
115
tests/test_learn/test_writer.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue