Plugin architecture for headroom learn + live traffic flush

Refactor headroom learn into a plugin architecture where each coding
agent (Claude Code, Codex, Gemini CLI) is a self-contained plugin
with scanner, writer, and detection logic. External plugins can
register via the headroom.learn_plugin entry point.

- Add LearnPlugin ABC (base.py) and plugin registry (registry.py)
- Move scanners from monolithic scanner.py into plugins/ directory
- Extract shared error classification and tool name map (_shared.py)
- Add GeminiScanner for Google Gemini CLI session parsing
- CLI uses dynamic agent detection via registry (no hardcoded choices)
- All existing imports preserved via backwards-compat re-exports
- Wire agent_type through wrap → proxy → TrafficLearner
- Flush learned patterns to correct .md file at proxy shutdown
- Fix shutdown queue drain bug (patterns were lost on exit)
- 97 tests pass (84 existing + 13 new registry/plugin tests)

headroom/cli/learn.py

@@ -8,59 +8,53 @@ from typing import TYPE_CHECKING
 import click
 
 if TYPE_CHECKING:
-    from ..learn.scanner import ConversationScanner
-    from ..learn.writer import ContextWriter
+    from ..learn.base import LearnPlugin
 
 from .main import main
 
-_AGENT_HELP = """Which coding agent to analyze. Auto-detects by default.
-
-\b
-Supported agents:
-  claude   Claude Code (~/.claude/)
-  codex    OpenAI Codex CLI (~/.codex/)
-  gemini   Google Gemini CLI (~/.gemini/)
-  auto     Auto-detect (check all, default)
-"""
-
 
-def _get_scanner_writer(agent: str) -> tuple[ConversationScanner, ContextWriter]:
-    """Get the appropriate scanner and writer for an agent type."""
-    from ..learn.scanner import ClaudeCodeScanner, CodexScanner
-    from ..learn.writer import ClaudeCodeWriter, CodexWriter
+class _AgentChoice(click.ParamType):
+    """Dynamic Click type that validates against the plugin registry."""
 
-    scanners = {
-        "claude": (ClaudeCodeScanner, ClaudeCodeWriter),
-        "codex": (CodexScanner, CodexWriter),
-        # Gemini scanner not yet implemented (protobuf sessions)
-        # Cursor scanner not yet implemented (SQLite blobs)
-    }
+    name = "agent"
 
-    if agent in scanners:
-        scanner_cls, writer_cls = scanners[agent]
-        return scanner_cls(), writer_cls()
+    def get_metavar(self, param: click.Parameter, ctx: click.Context | None = None) -> str | None:
+        return "[auto|<agent>]"
 
-    raise click.BadParameter(f"Unknown agent: {agent}. Supported: claude, codex")
+    def convert(
+        self,
+        value: str,
+        param: click.Parameter | None,
+        ctx: click.Context | None,
+    ) -> str:
+        if value == "auto":
+            return value
+        from ..learn.registry import get_registry
 
+        reg = get_registry()
+        if value.lower() not in reg:
+            available = ", ".join(sorted(reg.keys()))
+            self.fail(f"Unknown agent: {value}. Available: auto, {available}", param, ctx)
+        return value.lower()
 
-def _auto_detect_agents() -> list[tuple[str, ConversationScanner, ContextWriter]]:
-    """Auto-detect which agents have data on this machine."""
-    from ..learn.scanner import ClaudeCodeScanner, CodexScanner
-    from ..learn.writer import ClaudeCodeWriter, CodexWriter
+    def shell_complete(
+        self,
+        ctx: click.Context,
+        param: click.Parameter,
+        incomplete: str,
+    ) -> list[click.shell_completion.CompletionItem]:
+        from ..learn.registry import available_agent_names
 
-    agents: list[tuple[str, ConversationScanner, ContextWriter]] = []
+        names = ["auto"] + available_agent_names()
+        return [click.shell_completion.CompletionItem(n) for n in names if n.startswith(incomplete)]
 
-    # Claude Code
-    claude_dir = Path.home() / ".claude" / "projects"
-    if claude_dir.exists() and any(claude_dir.iterdir()):
-        agents.append(("claude", ClaudeCodeScanner(), ClaudeCodeWriter()))
 
-    # Codex
-    codex_dir = Path.home() / ".codex" / "sessions"
-    if codex_dir.exists() and any(codex_dir.glob("*.json")):
-        agents.append(("codex", CodexScanner(), CodexWriter()))
+_AGENT_HELP = """Which coding agent to analyze. Auto-detects by default.
 
-    return agents
+\b
+Built-in: claude, codex, gemini.
+External plugins register via 'headroom.learn_plugin' entry point.
+Use 'auto' (default) to scan all detected agents."""
 
 
 @main.command()
@@ -85,7 +79,7 @@ def _auto_detect_agents() -> list[tuple[str, ConversationScanner, ContextWriter]
 )
 @click.option(
     "--agent",
-    type=click.Choice(["auto", "claude", "codex", "gemini"], case_sensitive=False),
+    type=_AgentChoice(),
     default="auto",
     help=_AGENT_HELP,
 )
@@ -109,8 +103,9 @@ def learn(
     (wrong paths, missing modules, stubborn retries) and generates context
     that prevents them from recurring.
 
-    Supports multiple coding agents: Claude Code, Codex, Gemini CLI.
-    Uses LiteLLM for provider-agnostic LLM access (100+ models).
+    Supports multiple coding agents via a plugin architecture. Built-in
+    support for Claude Code, Codex, and Gemini CLI. External plugins can
+    be installed via pip (entry point: headroom.learn_plugin).
 
     \b
     Examples:
@@ -121,6 +116,7 @@ def learn(
         headroom learn --agent codex --all    # Analyze all Codex sessions
     """
     from ..learn.analyzer import SessionAnalyzer, _detect_default_model
+    from ..learn.registry import auto_detect_plugins, get_plugin
 
     # Resolve model early to fail fast with a clear message
     try:
@@ -132,15 +128,18 @@ def learn(
     analyzer = SessionAnalyzer(model=resolved_model)
 
     # Determine which agents to scan
+    agent_configs: list[tuple[str, LearnPlugin]] = []
+
     if agent == "auto":
-        agent_configs = _auto_detect_agents()
-        if not agent_configs:
-            click.echo("No coding agent data found. Checked: ~/.claude/, ~/.codex/")
+        detected = auto_detect_plugins()
+        if not detected:
+            click.echo("No coding agent data found.")
             return
-        click.echo(f"Detected agents: {', '.join(name for name, _, _ in agent_configs)}")
+        click.echo(f"Detected agents: {', '.join(p.display_name for p in detected)}")
+        agent_configs = [(p.name, p) for p in detected]
     else:
-        scanner, writer = _get_scanner_writer(agent)
-        agent_configs = [(agent, scanner, writer)]
+        selected = get_plugin(agent)
+        agent_configs = [(selected.name, selected)]
 
     total_projects = 0
     total_failures = 0
@@ -148,11 +147,12 @@ def learn(
     matched_projects = 0
     available_projects: list[tuple[str, Path]] = []
 
-    for agent_name, scanner, writer in agent_configs:
-        all_projects = scanner.discover_projects()
+    for agent_name, plugin in agent_configs:
+        writer = plugin.create_writer()
+        all_projects = plugin.discover_projects()
         if not all_projects:
             continue
-        available_projects.extend((agent_name, p.project_path) for p in all_projects)
+        available_projects.extend((agent_name, proj.project_path) for proj in all_projects)
 
         # Filter to target project(s)
         if analyze_all:
@@ -174,8 +174,8 @@ def learn(
                 click.echo(f"No {agent_name} project data found for {cwd}")
                 click.echo("Try: headroom learn --all  or  headroom learn --project <path>")
                 click.echo(f"\nAvailable {agent_name} projects:")
-                for p in all_projects[:10]:
-                    click.echo(f"  {p.name:30s} {p.project_path}")
+                for proj_info in all_projects[:10]:
+                    click.echo(f"  {proj_info.name:30s} {proj_info.project_path}")
                 return
 
         for proj in targets:
@@ -185,7 +185,7 @@ def learn(
             click.echo(f"Path: {proj.project_path}")
             click.echo(f"{'=' * 60}")
 
-            sessions = scanner.scan_project(proj)
+            sessions = plugin.scan_project(proj)
             if not sessions:
                 click.echo("  No conversation data found.")
                 continue

headroom/cli/proxy.py

@@ -269,6 +269,7 @@ def proxy(
         memory_top_k=memory_top_k,
         # Traffic Learning: only with --learn, never with --no-learn
         traffic_learning_enabled=learn and not no_learn,
+        traffic_learning_agent_type=os.environ.get("HEADROOM_AGENT_TYPE", "unknown"),
         # Backend (Anthropic direct, Bedrock, LiteLLM, or any-llm)
         backend=backend,
         bedrock_region=bedrock_region or region,

headroom/cli/wrap.py

@@ -60,6 +60,7 @@ def _start_proxy(
     port: int,
     *,
     learn: bool = False,
+    agent_type: str = "unknown",
     backend: str | None = None,
     anyllm_provider: str | None = None,
     region: str | None = None,
@@ -101,6 +102,10 @@ def _start_proxy(
     proxy_env = os.environ.copy()
     proxy_env["PYTHONIOENCODING"] = "utf-8"
 
+    # Tell the proxy which agent is being wrapped (for traffic learning output)
+    if agent_type != "unknown":
+        proxy_env["HEADROOM_AGENT_TYPE"] = agent_type
+
     proc = subprocess.Popen(
         cmd,
         stdout=log_file,
@@ -313,6 +318,7 @@ def _ensure_proxy(
     no_proxy: bool,
     *,
     learn: bool = False,
+    agent_type: str = "unknown",
     backend: str | None = None,
     anyllm_provider: str | None = None,
     region: str | None = None,
@@ -328,6 +334,7 @@ def _ensure_proxy(
                 proc = _start_proxy(
                     port,
                     learn=learn,
+                    agent_type=agent_type,
                     backend=backend,
                     anyllm_provider=anyllm_provider,
                     region=region,
@@ -392,6 +399,7 @@ def _launch_tool(
     env_vars_display: list[str],
     *,
     learn: bool = False,
+    agent_type: str = "unknown",
     backend: str | None = None,
     anyllm_provider: str | None = None,
     region: str | None = None,
@@ -414,6 +422,7 @@ def _launch_tool(
             port,
             no_proxy,
             learn=learn,
+            agent_type=agent_type,
             backend=backend,
             anyllm_provider=anyllm_provider,
             region=region,
@@ -713,7 +722,7 @@ def claude(
         click.echo("  ╚═══════════════════════════════════════════════╝")
         click.echo()
 
-        proxy_holder[0] = _ensure_proxy(port, no_proxy, learn=learn)
+        proxy_holder[0] = _ensure_proxy(port, no_proxy, learn=learn, agent_type="claude")
 
         if not no_rtk:
             click.echo("  Setting up rtk...")
@@ -832,6 +841,7 @@ def codex(
         tool_label="CODEX",
         env_vars_display=[f"OPENAI_BASE_URL=http://127.0.0.1:{port}/v1"],
         learn=learn,
+        agent_type="codex",
         backend=backend,
         anyllm_provider=anyllm_provider,
         region=region,
@@ -912,6 +922,7 @@ def aider(
             f"ANTHROPIC_BASE_URL=http://127.0.0.1:{port}",
         ],
         learn=learn,
+        agent_type="aider",
         backend=backend,
         anyllm_provider=anyllm_provider,
         region=region,
@@ -961,7 +972,7 @@ def cursor(port: int, no_rtk: bool, no_proxy: bool, learn: bool, verbose: bool)
         click.echo("  ╚═══════════════════════════════════════════════╝")
         click.echo()
 
-        proxy_holder[0] = _ensure_proxy(port, no_proxy, learn=learn)
+        proxy_holder[0] = _ensure_proxy(port, no_proxy, learn=learn, agent_type="cursor")
 
         # Setup rtk for Cursor (binary + .cursorrules instructions)
         if not no_rtk:

headroom/learn/__init__.py

@@ -1,12 +1,14 @@
 """Headroom Learn — offline session learning for coding agents.
 
-Analyzes conversation logs using Sonnet 4.6 to extract actionable patterns
-and generates context (CLAUDE.md, MEMORY.md, AGENTS.md, etc.) that prevents
-future token waste.
+Analyzes conversation logs using an LLM to extract actionable patterns
+and generates context (CLAUDE.md, AGENTS.md, GEMINI.md, etc.) that
+prevents future token waste.
 
-Architecture:
-    Scanner (adapter)  →  Analyzer (LLM)  →  Writer (adapter)
-    ├── ClaudeCodeScanner   SessionAnalyzer     ├── ClaudeCodeWriter
-    └── CodexScanner        (Sonnet 4.6)        ├── CodexWriter
-                                                └── GeminiWriter
+Plugin architecture:
+    plugins/claude.py  ─┐
+    plugins/codex.py   ─┤→  Analyzer (LLM)  →  Writer (adapter)
+    plugins/gemini.py  ─┘
+
+Built-in plugins are auto-discovered from headroom.learn.plugins.*.
+External plugins register via the ``headroom.learn_plugin`` entry point.
 """

headroom/learn/_shared.py

@@ -0,0 +1,146 @@
+"""Shared utilities for headroom learn plugins.
+
+Error classification, tool name normalization, and other helpers
+used across all scanner plugins.
+"""
+
+from __future__ import annotations
+
+import re
+
+from .models import ErrorCategory
+
+# =============================================================================
+# Error Classification
+# =============================================================================
+
+# Patterns checked in order — first match wins
+_ERROR_PATTERNS: list[tuple[re.Pattern[str], 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)
+
+
+# =============================================================================
+# Tool Name Normalization
+# =============================================================================
+
+# Consolidated mapping from all agent-specific tool names to the cross-agent schema.
+# Plugins can use normalize_tool_name() or extend this map for custom tools.
+_TOOL_NAME_MAP: dict[str, str] = {
+    # Shell / command execution
+    "shell": "Bash",
+    "run_shell_command": "Bash",
+    "execute_command": "Bash",
+    "exec_command": "Bash",
+    "terminal": "Bash",
+    "run_command": "Bash",
+    "run_terminal_command": "Bash",
+    # File reading
+    "read_file": "Read",
+    "read_many_files": "Read",
+    "readfile": "Read",
+    "view_file": "Read",
+    "cat": "Read",
+    # File writing
+    "write_file": "Write",
+    "write_new_file": "Write",
+    "create_file": "Write",
+    "writefile": "Write",
+    # File editing
+    "edit_file": "Edit",
+    "replace_in_file": "Edit",
+    "editfile": "Edit",
+    "apply_diff": "Edit",
+    # File search / glob
+    "search_files": "Glob",
+    "find_files": "Glob",
+    "glob": "Glob",
+    "list_directory": "Glob",
+    "list_dir": "Glob",
+    # Text search / grep
+    "grep": "Grep",
+    "search_text": "Grep",
+    "search_code": "Grep",
+    "codebase_search": "Grep",
+    # Web
+    "browser": "WebFetch",
+    "web_search": "WebSearch",
+}
+
+
+def normalize_tool_name(name: str) -> str:
+    """Map agent-specific tool names to the cross-agent schema.
+
+    Looks up the name (case-insensitive) in the shared tool name map.
+    Returns the original name if no mapping exists.
+    """
+    return _TOOL_NAME_MAP.get(name.lower(), _TOOL_NAME_MAP.get(name, name))

headroom/learn/base.py

@@ -0,0 +1,111 @@
+"""Base class for headroom learn plugins.
+
+Each coding agent (Claude Code, Codex, Gemini, Cursor, etc.) implements
+a LearnPlugin that bundles scanning, writing, and detection into one unit.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from .models import ProjectInfo, SessionData
+from .writer import ContextWriter
+
+
+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."""
+        ...
+
+
+class LearnPlugin(ABC):
+    """A self-contained learn plugin for a single coding agent.
+
+    Bundles identity, detection, scanning, and writer creation.
+    Plugins are discovered automatically from headroom.learn.plugins.*
+    or via ``headroom.learn_plugin`` entry points for external packages.
+
+    Example::
+
+        class MyAgentPlugin(LearnPlugin, ConversationScanner):
+            @property
+            def name(self) -> str:
+                return "myagent"
+
+            @property
+            def display_name(self) -> str:
+                return "My Agent"
+
+            def detect(self) -> bool:
+                return Path("~/.myagent/sessions").expanduser().exists()
+
+            def discover_projects(self) -> list[ProjectInfo]: ...
+            def scan_project(self, project: ProjectInfo) -> list[SessionData]: ...
+
+            def create_writer(self) -> ContextWriter:
+                from headroom.learn.writer import GeminiWriter
+                return GeminiWriter()  # or a custom writer
+
+        # Module-level instance for auto-discovery
+        plugin = MyAgentPlugin()
+    """
+
+    # --- Identity ---
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Short lowercase identifier used in CLI (e.g., 'claude', 'cursor')."""
+        ...
+
+    @property
+    @abstractmethod
+    def display_name(self) -> str:
+        """Human-readable name (e.g., 'Claude Code', 'Cursor')."""
+        ...
+
+    @property
+    def description(self) -> str:
+        """One-line description for --help output."""
+        return f"{self.display_name} coding agent"
+
+    # --- Detection ---
+
+    @abstractmethod
+    def detect(self) -> bool:
+        """Return True if this agent has data on the current machine.
+
+        Called during auto-detection. Must be cheap (stat checks only, no I/O).
+        """
+        ...
+
+    # --- Scanning ---
+
+    @abstractmethod
+    def discover_projects(self) -> list[ProjectInfo]:
+        """Discover all projects with conversation data for this agent."""
+        ...
+
+    @abstractmethod
+    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
+        """Scan all sessions for a project, returning normalized data."""
+        ...
+
+    # --- Writing ---
+
+    @abstractmethod
+    def create_writer(self) -> ContextWriter:
+        """Return the appropriate ContextWriter for this agent."""
+        ...

headroom/learn/plugins/__init__.py

@@ -0,0 +1,5 @@
+"""Built-in learn plugins for headroom.
+
+Each module in this package exposes a ``plugin`` attribute (a ``LearnPlugin``
+instance) that is auto-discovered by the plugin registry.
+"""

headroom/learn/plugins/claude.py

@@ -0,0 +1,399 @@
+"""Claude Code plugin for headroom learn.
+
+Reads conversation logs from ~/.claude/projects/ (JSONL format).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from pathlib import Path
+
+from .._shared import classify_error, is_error_content
+from ..base import ConversationScanner, LearnPlugin
+from ..models import (
+    ErrorCategory,
+    ProjectInfo,
+    SessionData,
+    SessionEvent,
+    ToolCall,
+)
+from ..writer import ClaudeCodeWriter, ContextWriter
+
+logger = logging.getLogger(__name__)
+
+
+class ClaudeCodePlugin(LearnPlugin, 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"
+
+    # --- LearnPlugin identity ---
+
+    @property
+    def name(self) -> str:
+        return "claude"
+
+    @property
+    def display_name(self) -> str:
+        return "Claude Code"
+
+    @property
+    def description(self) -> str:
+        return "Claude Code (~/.claude/)"
+
+    def detect(self) -> bool:
+        return self.projects_dir.exists() and any(self.projects_dir.iterdir())
+
+    def create_writer(self) -> ContextWriter:
+        return ClaudeCodeWriter()
+
+    # --- ConversationScanner interface ---
+
+    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
+
+            project_path = _decode_project_path(entry.name)
+            if project_path is None:
+                fallback_parts = entry.name[1:].split("-")
+                if len(fallback_parts[0]) == 1 and fallback_parts[0].isalpha():
+                    drive = fallback_parts[0].upper()
+                    project_path = Path(f"{drive}:\\" + "\\".join(fallback_parts[1:]))
+                else:
+                    project_path = Path("/" + entry.name[1:].replace("-", "/"))
+
+            name = project_path.name if project_path != Path("/") else entry.name
+
+            context_file = None
+            if project_path.exists():
+                claude_md = project_path / "CLAUDE.md"
+                if claude_md.exists():
+                    context_file = claude_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
+
+            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 = []
+        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]] = {}
+        tool_calls: list[ToolCall] = []
+        events: list[SessionEvent] = []
+        total_input_tokens = 0
+        total_output_tokens = 0
+        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", "")
+                    ts = d.get("timestamp", None)
+
+                    if line_type == "assistant":
+                        self._extract_tool_uses(d, tool_uses)
+                        usage = d.get("message", {}).get("usage", {})
+                        total_input_tokens += usage.get("input_tokens", 0)
+                        total_input_tokens += usage.get("cache_read_input_tokens", 0)
+                        total_input_tokens += usage.get("cache_creation_input_tokens", 0)
+                        total_output_tokens += usage.get("output_tokens", 0)
+                    elif line_type == "user":
+                        self._extract_tool_results(d, tool_uses, tool_calls, events, msg_index, ts)
+                        self._extract_user_events(d, events, msg_index, ts)
+
+        except (OSError, UnicodeDecodeError) as e:
+            logger.debug("Failed to read %s: %s", jsonl_path, e)
+            return None
+
+        for tc in tool_calls:
+            if not any(e.type == "tool_call" and e.tool_call is tc for e in events):
+                events.append(SessionEvent(type="tool_call", msg_index=tc.msg_index, tool_call=tc))
+        events.sort(key=lambda e: e.msg_index)
+
+        return SessionData(
+            session_id=session_id,
+            tool_calls=tool_calls,
+            events=events,
+            total_input_tokens=total_input_tokens,
+            total_output_tokens=total_output_tokens,
+        )
+
+    def _extract_tool_uses(self, d: dict, tool_uses: dict[str, tuple[str, dict]]) -> None:
+        """Extract tool_use blocks from an assistant message."""
+        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],
+        events: list[SessionEvent],
+        msg_index: int,
+        timestamp: str | None = None,
+    ) -> None:
+        """Extract tool_result blocks from a user message and match to tool_uses."""
+        msg = d.get("message", {})
+        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)
+
+            if tc_id not in tool_uses:
+                continue
+
+            name, inp = tool_uses[tc_id]
+
+            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
+
+            tc = ToolCall(
+                name=name,
+                tool_call_id=tc_id,
+                input_data=inp,
+                output=result_content,
+                is_error=is_err,
+                error_category=error_cat,
+                msg_index=msg_index,
+                output_bytes=len(result_content.encode("utf-8")),
+            )
+            tool_calls.append(tc)
+            events.append(
+                SessionEvent(
+                    type="tool_call", msg_index=msg_index, timestamp=timestamp, tool_call=tc
+                )
+            )
+
+            if name in ("Agent", "agent"):
+                tool_result_meta = d.get("toolUseResult", {})
+                if isinstance(tool_result_meta, dict):
+                    events.append(
+                        SessionEvent(
+                            type="agent_summary",
+                            msg_index=msg_index,
+                            timestamp=timestamp,
+                            agent_id=tool_result_meta.get("agentId", ""),
+                            agent_tool_count=tool_result_meta.get("totalToolUseCount", 0),
+                            agent_tokens=tool_result_meta.get("totalTokens", 0),
+                            agent_duration_ms=tool_result_meta.get("totalDurationMs", 0),
+                            agent_prompt=tool_result_meta.get("prompt", "")[:200],
+                        )
+                    )
+
+    def _extract_user_events(
+        self,
+        d: dict,
+        events: list[SessionEvent],
+        msg_index: int,
+        timestamp: str | None = None,
+    ) -> None:
+        """Extract user text messages and interruptions from a user line."""
+        msg = d.get("message", {})
+        content = msg.get("content", "")
+
+        if isinstance(content, str) and content.strip():
+            events.append(
+                SessionEvent(
+                    type="user_message",
+                    msg_index=msg_index,
+                    timestamp=timestamp,
+                    text=content[:500],
+                )
+            )
+            return
+
+        if isinstance(content, list):
+            for block in content:
+                if not isinstance(block, dict):
+                    continue
+                if block.get("type") == "text":
+                    text = block.get("text", "")
+                    if "[Request interrupted by user" in text:
+                        events.append(
+                            SessionEvent(
+                                type="interruption",
+                                msg_index=msg_index,
+                                timestamp=timestamp,
+                                text=text[:200],
+                            )
+                        )
+
+
+# =============================================================================
+# Path Decode Helpers (Claude Code specific)
+# =============================================================================
+
+
+def _decode_project_path(escaped_name: str) -> Path | None:
+    """Decode a Claude Code escaped project path."""
+    if not escaped_name.startswith("-"):
+        return None
+
+    parts = escaped_name[1:].split("-")
+    if len(parts) < 2:
+        return None
+
+    if len(parts[0]) == 1 and parts[0].isalpha():
+        drive = parts[0].upper()
+        win_path = Path(f"{drive}:\\" + "\\".join(parts[1:]))
+        if win_path.exists():
+            return win_path
+        win_base = Path(f"{drive}:\\{parts[1]}") if len(parts) > 1 else win_path
+        if win_base.exists() and len(parts) > 2:
+            result = _greedy_path_decode(win_base, parts[2:])
+            if result:
+                return result
+
+    simple = Path("/" + escaped_name[1:].replace("-", "/"))
+    if simple.exists():
+        return simple
+
+    if len(parts) < 3:
+        return None
+
+    if parts[0] == "Users" and len(parts) > 2:
+        base = Path(f"/{parts[0]}/{parts[1]}")
+        remaining = parts[2:]
+        return _greedy_path_decode(base, remaining)
+
+    if parts[0] == "home" 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 using real child directories."""
+    if not parts:
+        return base if base.exists() else None
+
+    if not base.exists() or not base.is_dir():
+        return None
+
+    try:
+        children = sorted(child for child in base.iterdir() if child.is_dir())
+    except OSError:
+        return None
+
+    for child in children:
+        for tokenization in _component_tokenizations(child.name):
+            n_tokens = len(tokenization)
+            if parts[:n_tokens] != tokenization:
+                continue
+
+            result = _greedy_path_decode(child, parts[n_tokens:])
+            if result:
+                return result
+
+    return None
+
+
+def _component_tokenizations(component: str) -> list[list[str]]:
+    """Return possible escaped token sequences for a real path component."""
+    tokenizations: list[list[str]] = []
+    seen: set[tuple[str, ...]] = set()
+
+    def add(tokens: list[str]) -> None:
+        key = tuple(tokens)
+        if tokens and key not in seen:
+            seen.add(key)
+            tokenizations.append(tokens)
+
+    add([component])
+
+    for separator in ("-", ".", None):
+        if separator is None:
+            tokens = [token for token in re.split(r"[-.]", component) if token]
+        else:
+            tokens = [token for token in component.split(separator) if token]
+        add(tokens)
+
+    if component.startswith(".") and len(component) > 1:
+        hidden_component = component[1:]
+        add(["", hidden_component])
+        for separator in ("-", ".", None):
+            if separator is None:
+                tokens = [token for token in re.split(r"[-.]", hidden_component) if token]
+            else:
+                tokens = [token for token in hidden_component.split(separator) if token]
+            add(["", *tokens])
+
+    return tokenizations
+
+
+# Module-level instance for auto-discovery by the plugin registry
+plugin = ClaudeCodePlugin()

headroom/learn/plugins/codex.py

@@ -0,0 +1,315 @@
+"""OpenAI Codex CLI plugin for headroom learn.
+
+Reads session logs from ~/.codex/sessions/ (JSON and JSONL formats).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+from .._shared import classify_error, is_error_content, normalize_tool_name
+from ..base import ConversationScanner, LearnPlugin
+from ..models import (
+    ErrorCategory,
+    ProjectInfo,
+    SessionData,
+    ToolCall,
+)
+from ..writer import CodexWriter, ContextWriter
+
+logger = logging.getLogger(__name__)
+
+
+class CodexPlugin(LearnPlugin, ConversationScanner):
+    """Reads OpenAI Codex CLI session logs from ~/.codex/sessions/.
+
+    Codex stores sessions as JSON files with:
+    - session.id, session.timestamp, session.instructions
+    - items[]: array of message/function_call/function_call_output/reasoning objects
+
+    function_call items have: name, call_id, arguments (JSON string)
+    function_call_output items have: call_id, output (string or JSON string)
+    """
+
+    def __init__(self, codex_dir: Path | None = None):
+        self.codex_dir = codex_dir or Path.home() / ".codex"
+        self.sessions_dir = self.codex_dir / "sessions"
+
+    # --- LearnPlugin identity ---
+
+    @property
+    def name(self) -> str:
+        return "codex"
+
+    @property
+    def display_name(self) -> str:
+        return "OpenAI Codex CLI"
+
+    @property
+    def description(self) -> str:
+        return "OpenAI Codex CLI (~/.codex/)"
+
+    def detect(self) -> bool:
+        if not self.sessions_dir.exists():
+            return False
+        return bool(
+            any(self.sessions_dir.rglob("*.json")) or any(self.sessions_dir.rglob("*.jsonl"))
+        )
+
+    def create_writer(self) -> ContextWriter:
+        return CodexWriter()
+
+    # --- ConversationScanner interface ---
+
+    def _iter_session_files(self, root: Path | None = None) -> list[Path]:
+        """Return all known Codex session files, including nested rollouts."""
+        search_root = root or self.sessions_dir
+        session_files = list(search_root.rglob("*.json")) + list(search_root.rglob("*.jsonl"))
+        return sorted(path for path in session_files if path.is_file())
+
+    def discover_projects(self) -> list[ProjectInfo]:
+        """Codex doesn't organize by project — return a single 'codex' project."""
+        if not self.sessions_dir.exists():
+            return []
+
+        session_files = self._iter_session_files()
+        if not session_files:
+            return []
+
+        agents_md = self.codex_dir / "AGENTS.md"
+        instructions_md = self.codex_dir / "instructions.md"
+
+        return [
+            ProjectInfo(
+                name="codex",
+                project_path=Path.cwd(),
+                data_path=self.sessions_dir,
+                context_file=agents_md if agents_md.exists() else None,
+                memory_file=instructions_md if instructions_md.exists() else None,
+            )
+        ]
+
+    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
+        """Scan all Codex session JSON files."""
+        sessions = []
+        for json_path in self._iter_session_files(project.data_path):
+            session = self._scan_session(json_path)
+            if session and session.tool_calls:
+                sessions.append(session)
+        return sessions
+
+    def _scan_session(self, json_path: Path) -> SessionData | None:
+        """Parse a single Codex session file."""
+        if json_path.suffix == ".jsonl":
+            return self._scan_jsonl_session(json_path)
+        return self._scan_json_session(json_path)
+
+    def _scan_json_session(self, json_path: Path) -> SessionData | None:
+        """Parse a single Codex session file."""
+        try:
+            with open(json_path) as f:
+                data = json.load(f)
+        except (OSError, json.JSONDecodeError) as e:
+            logger.debug("Failed to read Codex session %s: %s", json_path, e)
+            return None
+
+        session_info = data.get("session", {})
+        session_id = session_info.get("id", json_path.stem)
+        items = data.get("items", [])
+
+        if not items:
+            return None
+
+        func_calls: dict[str, tuple[str, dict]] = {}
+        tool_calls: list[ToolCall] = []
+        msg_index = 0
+
+        for item in items:
+            msg_index += 1
+            item_type = item.get("type", "")
+
+            if item_type == "function_call":
+                call_id = item.get("call_id", "")
+                name = item.get("name", "")
+                raw_args = item.get("arguments", "")
+                if isinstance(raw_args, str):
+                    try:
+                        parsed = json.loads(raw_args)
+                    except (json.JSONDecodeError, TypeError):
+                        parsed = {"raw": raw_args}
+                elif isinstance(raw_args, dict):
+                    parsed = raw_args
+                else:
+                    parsed = {"raw": str(raw_args)}
+
+                # Codex-specific: extract command from shell args
+                if name == "shell" and "command" in parsed:
+                    cmd = parsed["command"]
+                    if isinstance(cmd, list):
+                        parsed["command"] = cmd[-1] if cmd else ""
+                    name = "Bash"
+                else:
+                    name = normalize_tool_name(name)
+
+                if call_id:
+                    func_calls[call_id] = (name, parsed)
+
+            elif item_type == "function_call_output":
+                call_id = item.get("call_id", "")
+                output_raw = item.get("output", "")
+
+                if call_id not in func_calls:
+                    continue
+
+                name, inp = func_calls[call_id]
+                result_content = _parse_codex_output(output_raw)
+
+                is_err = is_error_content(result_content)
+                error_cat = classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
+
+                tool_calls.append(
+                    ToolCall(
+                        name=name,
+                        tool_call_id=call_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")),
+                    )
+                )
+
+        return SessionData(session_id=session_id, tool_calls=tool_calls)
+
+    def _scan_jsonl_session(self, jsonl_path: Path) -> SessionData | None:
+        """Parse a modern Codex rollout session stored as JSONL."""
+        session_id = jsonl_path.stem
+        func_calls: dict[str, tuple[str, dict]] = {}
+        tool_calls: list[ToolCall] = []
+        msg_index = 0
+
+        try:
+            with open(jsonl_path) as f:
+                for line in f:
+                    try:
+                        entry = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+
+                    if entry.get("type") == "session_meta":
+                        payload = entry.get("payload", {})
+                        if isinstance(payload, dict):
+                            session_id = payload.get("id", session_id)
+                        continue
+
+                    if entry.get("type") != "response_item":
+                        continue
+
+                    payload = entry.get("payload", {})
+                    if not isinstance(payload, dict):
+                        continue
+
+                    msg_index += 1
+                    item_type = payload.get("type", "")
+
+                    if item_type in ("function_call", "custom_tool_call"):
+                        call_id = payload.get("call_id", "")
+                        name = payload.get("name", "")
+                        parsed = _parse_codex_arguments(payload)
+                        name, parsed = _normalize_codex_tool(name, parsed)
+                        if call_id and name:
+                            func_calls[call_id] = (name, parsed)
+                        continue
+
+                    if item_type not in ("function_call_output", "custom_tool_call_output"):
+                        continue
+
+                    call_id = payload.get("call_id", "")
+                    if call_id not in func_calls:
+                        continue
+
+                    name, inp = func_calls[call_id]
+                    result_content = _parse_codex_output(payload.get("output", ""))
+                    is_err = is_error_content(result_content)
+                    error_cat = classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
+
+                    tool_calls.append(
+                        ToolCall(
+                            name=name,
+                            tool_call_id=call_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")),
+                        )
+                    )
+
+        except OSError as e:
+            logger.debug("Failed to read Codex session %s: %s", jsonl_path, e)
+            return None
+
+        if not tool_calls:
+            return None
+
+        return SessionData(session_id=session_id, tool_calls=tool_calls)
+
+
+# =============================================================================
+# Codex-specific Helpers
+# =============================================================================
+
+
+def _parse_codex_arguments(payload: dict) -> dict:
+    """Parse arguments for either legacy or rollout Codex tool calls."""
+    raw_args = payload.get("arguments", payload.get("input", ""))
+    if isinstance(raw_args, str):
+        try:
+            parsed = json.loads(raw_args)
+            return parsed if isinstance(parsed, dict) else {"raw": raw_args}
+        except (json.JSONDecodeError, TypeError):
+            return {"raw": raw_args}
+    if isinstance(raw_args, dict):
+        return raw_args
+    return {"raw": str(raw_args)}
+
+
+def _normalize_codex_tool(name: str, parsed: dict) -> tuple[str, dict]:
+    """Normalize modern Codex tool names to the cross-agent schema."""
+    if name == "shell" and "command" in parsed:
+        cmd = parsed["command"]
+        if isinstance(cmd, list):
+            parsed["command"] = cmd[-1] if cmd else ""
+        return "Bash", parsed
+
+    if name == "exec_command" and "cmd" in parsed:
+        parsed = dict(parsed)
+        parsed["command"] = parsed.get("cmd", "")
+        return "Bash", parsed
+
+    return normalize_tool_name(name), parsed
+
+
+def _parse_codex_output(output_raw: object) -> str:
+    """Parse tool output from Codex rollout records."""
+    if isinstance(output_raw, str):
+        try:
+            parsed_out = json.loads(output_raw)
+        except (json.JSONDecodeError, TypeError):
+            return output_raw
+
+        if isinstance(parsed_out, dict):
+            if "output" in parsed_out:
+                return str(parsed_out["output"])
+            return json.dumps(parsed_out)
+        return output_raw
+
+    return str(output_raw)
+
+
+# Module-level instance for auto-discovery by the plugin registry
+plugin = CodexPlugin()

headroom/learn/plugins/gemini.py

@@ -0,0 +1,320 @@
+"""Google Gemini CLI plugin for headroom learn.
+
+Reads session logs from ~/.gemini/tmp/<project_hash>/chats/ (JSON and JSONL).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+from .._shared import classify_error, is_error_content, normalize_tool_name
+from ..base import ConversationScanner, LearnPlugin
+from ..models import (
+    ErrorCategory,
+    ProjectInfo,
+    SessionData,
+    SessionEvent,
+    ToolCall,
+)
+from ..writer import ContextWriter, GeminiWriter
+
+logger = logging.getLogger(__name__)
+
+
+class GeminiPlugin(LearnPlugin, ConversationScanner):
+    """Reads Google Gemini CLI session logs from ~/.gemini/tmp/<project>/chats/.
+
+    Gemini CLI stores sessions as JSON or JSONL files with messages in the
+    Gemini API format:
+    - role: "user" or "model"
+    - parts[]: array containing text, functionCall, or functionResponse objects
+
+    Tool calls use:
+    - functionCall: {name, args}  (in model messages)
+    - functionResponse: {name, response}  (in user messages)
+    """
+
+    def __init__(self, gemini_dir: Path | None = None):
+        self.gemini_dir = gemini_dir or Path.home() / ".gemini"
+        self.tmp_dir = self.gemini_dir / "tmp"
+
+    # --- LearnPlugin identity ---
+
+    @property
+    def name(self) -> str:
+        return "gemini"
+
+    @property
+    def display_name(self) -> str:
+        return "Google Gemini CLI"
+
+    @property
+    def description(self) -> str:
+        return "Google Gemini CLI (~/.gemini/)"
+
+    def detect(self) -> bool:
+        if not self.tmp_dir.exists():
+            return False
+        return bool(
+            any(self.tmp_dir.rglob("session-*.json")) or any(self.tmp_dir.rglob("session-*.jsonl"))
+        )
+
+    def create_writer(self) -> ContextWriter:
+        return GeminiWriter()
+
+    # --- ConversationScanner interface ---
+
+    def discover_projects(self) -> list[ProjectInfo]:
+        """Discover all projects with Gemini session data."""
+        if not self.tmp_dir.exists():
+            return []
+
+        projects = []
+        for project_dir in sorted(self.tmp_dir.iterdir()):
+            if not project_dir.is_dir():
+                continue
+
+            chats_dir = project_dir / "chats"
+            if not chats_dir.exists():
+                continue
+
+            session_files = list(chats_dir.glob("session-*.json")) + list(
+                chats_dir.glob("session-*.jsonl")
+            )
+            if not session_files:
+                continue
+
+            project_path = self._detect_project_path(session_files[0])
+
+            gemini_md = None
+            if project_path and project_path.exists():
+                candidate = project_path / "GEMINI.md"
+                if candidate.exists():
+                    gemini_md = candidate
+
+            projects.append(
+                ProjectInfo(
+                    name=project_path.name if project_path else project_dir.name,
+                    project_path=project_path or Path.cwd(),
+                    data_path=chats_dir,
+                    context_file=gemini_md,
+                    memory_file=None,
+                )
+            )
+
+        return projects
+
+    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
+        """Scan all Gemini session files for a project."""
+        sessions = []
+        session_files = sorted(project.data_path.glob("session-*.json")) + sorted(
+            project.data_path.glob("session-*.jsonl")
+        )
+
+        for session_path in session_files:
+            session = self._scan_session(session_path)
+            if session and session.tool_calls:
+                sessions.append(session)
+
+        return sessions
+
+    def _scan_session(self, session_path: Path) -> SessionData | None:
+        """Parse a single Gemini session file (JSON or JSONL)."""
+        if session_path.suffix == ".jsonl":
+            return self._scan_jsonl_session(session_path)
+        return self._scan_json_session(session_path)
+
+    def _scan_json_session(self, json_path: Path) -> SessionData | None:
+        """Parse a Gemini JSON session file."""
+        try:
+            with open(json_path) as f:
+                data = json.load(f)
+        except (OSError, json.JSONDecodeError) as e:
+            logger.debug("Failed to read Gemini session %s: %s", json_path, e)
+            return None
+
+        session_id = json_path.stem
+
+        if isinstance(data, list):
+            messages = data
+        elif isinstance(data, dict):
+            messages = data.get("messages", data.get("history", data.get("contents", []))) or []
+            session_id = str(data.get("id", data.get("session_id", session_id)))
+        else:
+            return None
+
+        return self._parse_messages(session_id, messages)
+
+    def _scan_jsonl_session(self, jsonl_path: Path) -> SessionData | None:
+        """Parse a Gemini JSONL session file."""
+        session_id = jsonl_path.stem
+        messages: list[dict] = []
+
+        try:
+            with open(jsonl_path) as f:
+                for line in f:
+                    try:
+                        entry = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+
+                    entry_type = entry.get("type", "")
+
+                    if entry_type == "session_metadata":
+                        session_id = entry.get("id", entry.get("session_id", session_id))
+                        continue
+
+                    role = entry.get("role", "")
+                    if not role:
+                        if entry_type in ("user", "gemini", "model"):
+                            role = "model" if entry_type == "gemini" else entry_type
+                        else:
+                            continue
+
+                    parts = entry.get("parts", [])
+                    if parts:
+                        messages.append({"role": role, "parts": parts})
+
+        except (OSError, UnicodeDecodeError) as e:
+            logger.debug("Failed to read Gemini session %s: %s", jsonl_path, e)
+            return None
+
+        return self._parse_messages(session_id, messages)
+
+    def _parse_messages(self, session_id: str, messages: list) -> SessionData | None:
+        """Parse Gemini API format messages into normalized SessionData."""
+        tool_calls_pending: dict[str, tuple[str, dict, int]] = {}
+        tool_calls: list[ToolCall] = []
+        events: list[SessionEvent] = []
+        msg_index = 0
+        total_input_tokens = 0
+        total_output_tokens = 0
+
+        for msg in messages:
+            if not isinstance(msg, dict):
+                continue
+
+            msg_index += 1
+            role = msg.get("role", "")
+            parts = msg.get("parts", [])
+
+            usage = msg.get("usageMetadata", msg.get("usage", {}))
+            if isinstance(usage, dict):
+                total_input_tokens += usage.get("promptTokenCount", 0)
+                total_input_tokens += usage.get("cachedContentTokenCount", 0)
+                total_output_tokens += usage.get("candidatesTokenCount", 0)
+                total_output_tokens += (
+                    usage.get("totalTokenCount", 0) - usage.get("promptTokenCount", 0)
+                    if usage.get("totalTokenCount")
+                    else 0
+                )
+
+            if not isinstance(parts, list):
+                continue
+
+            for part in parts:
+                if not isinstance(part, dict):
+                    continue
+
+                if role == "user" and "text" in part:
+                    text = part["text"]
+                    if isinstance(text, str) and text.strip():
+                        events.append(
+                            SessionEvent(
+                                type="user_message",
+                                msg_index=msg_index,
+                                text=text[:500],
+                            )
+                        )
+
+                if "functionCall" in part:
+                    fc = part["functionCall"]
+                    if isinstance(fc, dict):
+                        name = fc.get("name", "")
+                        args = fc.get("args", {})
+                        if not isinstance(args, dict):
+                            args = {}
+                        normalized_name = normalize_tool_name(name)
+                        if name:
+                            tool_calls_pending[name] = (normalized_name, args, msg_index)
+
+                if "functionResponse" in part:
+                    fr = part["functionResponse"]
+                    if isinstance(fr, dict):
+                        name = fr.get("name", "")
+                        response = fr.get("response", {})
+
+                        if isinstance(response, dict):
+                            result_content = response.get("output", response.get("result", ""))
+                            if not isinstance(result_content, str):
+                                result_content = json.dumps(response)
+                        elif isinstance(response, str):
+                            result_content = response
+                        else:
+                            result_content = str(response)
+
+                        if name in tool_calls_pending:
+                            normalized_name, args, call_idx = tool_calls_pending.pop(name)
+                        else:
+                            normalized_name = normalize_tool_name(name)
+                            args = {}
+                            call_idx = msg_index
+
+                        call_id = f"{session_id}_{call_idx}_{name}"
+                        is_err = is_error_content(result_content)
+                        error_cat = (
+                            classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
+                        )
+
+                        tc = ToolCall(
+                            name=normalized_name,
+                            tool_call_id=call_id,
+                            input_data=args,
+                            output=result_content,
+                            is_error=is_err,
+                            error_category=error_cat,
+                            msg_index=msg_index,
+                            output_bytes=len(result_content.encode("utf-8")),
+                        )
+                        tool_calls.append(tc)
+                        events.append(
+                            SessionEvent(
+                                type="tool_call",
+                                msg_index=msg_index,
+                                tool_call=tc,
+                            )
+                        )
+
+        events.sort(key=lambda e: e.msg_index)
+
+        return SessionData(
+            session_id=session_id,
+            tool_calls=tool_calls,
+            events=events,
+            total_input_tokens=total_input_tokens,
+            total_output_tokens=total_output_tokens,
+        )
+
+    def _detect_project_path(self, session_path: Path) -> Path | None:
+        """Try to detect the project path from a session file."""
+        try:
+            with open(session_path) as f:
+                data = json.load(f)
+        except (OSError, json.JSONDecodeError):
+            return None
+
+        if isinstance(data, dict):
+            project_path = data.get("projectPath", data.get("project_path", ""))
+            if project_path and Path(project_path).exists():
+                return Path(project_path)
+            cwd = data.get("cwd", data.get("workingDirectory", ""))
+            if cwd and Path(cwd).exists():
+                return Path(cwd)
+
+        return None
+
+
+# Module-level instance for auto-discovery by the plugin registry
+plugin = GeminiPlugin()

headroom/learn/registry.py

@@ -0,0 +1,107 @@
+"""Plugin registry for headroom learn.
+
+Discovers built-in plugins from headroom.learn.plugins.* and external
+plugins registered via the ``headroom.learn_plugin`` entry point group.
+
+Follows the same pattern as headroom.storage_backend (storage/__init__.py).
+"""
+
+from __future__ import annotations
+
+import importlib
+import logging
+import pkgutil
+
+from .base import LearnPlugin
+
+logger = logging.getLogger(__name__)
+
+_registry: dict[str, LearnPlugin] | None = None
+
+
+def _discover() -> dict[str, LearnPlugin]:
+    """Discover all built-in and external plugins."""
+    plugins: dict[str, LearnPlugin] = {}
+
+    # 1. Built-in: scan headroom.learn.plugins.* submodules
+    from headroom.learn import plugins as plugins_pkg
+
+    for _, mod_name, _ in pkgutil.iter_modules(plugins_pkg.__path__):
+        try:
+            mod = importlib.import_module(f"headroom.learn.plugins.{mod_name}")
+            if hasattr(mod, "plugin"):
+                p = mod.plugin
+                if isinstance(p, LearnPlugin):
+                    plugins[p.name] = p
+                    logger.debug("Loaded built-in learn plugin: %s", p.name)
+        except Exception:
+            logger.debug("Failed to load built-in plugin: %s", mod_name, exc_info=True)
+
+    # 2. External: entry_points(group="headroom.learn_plugin")
+    try:
+        from importlib.metadata import entry_points
+
+        for ep in entry_points(group="headroom.learn_plugin"):
+            try:
+                obj = ep.load()
+                # Support both instances and factory callables
+                if isinstance(obj, LearnPlugin):
+                    p = obj
+                elif callable(obj):
+                    p = obj()
+                else:
+                    logger.warning("Learn plugin %s is not a LearnPlugin instance", ep.name)
+                    continue
+
+                if isinstance(p, LearnPlugin):
+                    plugins[p.name] = p  # External overrides built-in on name collision
+                    logger.debug("Loaded external learn plugin: %s (%s)", p.name, ep.name)
+            except Exception:
+                logger.warning("Failed to load external learn plugin: %s", ep.name, exc_info=True)
+    except Exception:
+        logger.debug("Entry point discovery failed", exc_info=True)
+
+    return plugins
+
+
+def get_registry() -> dict[str, LearnPlugin]:
+    """Get the plugin registry, discovering plugins on first call.
+
+    Returns a name → LearnPlugin mapping of all available plugins.
+    """
+    global _registry
+    if _registry is None:
+        _registry = _discover()
+    return _registry
+
+
+def get_plugin(name: str) -> LearnPlugin:
+    """Look up a plugin by name.
+
+    Raises KeyError with a helpful message if not found.
+    """
+    reg = get_registry()
+    if name not in reg:
+        available = ", ".join(sorted(reg.keys()))
+        raise KeyError(f"Unknown agent: {name!r}. Available: {available}")
+    return reg[name]
+
+
+def auto_detect_plugins() -> list[LearnPlugin]:
+    """Return plugins that have data on the current machine.
+
+    Calls ``detect()`` on each registered plugin and filters to those
+    that return True.
+    """
+    return [p for p in get_registry().values() if p.detect()]
+
+
+def available_agent_names() -> list[str]:
+    """Return sorted list of all registered agent names."""
+    return sorted(get_registry().keys())
+
+
+def reset_registry() -> None:
+    """Clear the registry cache. Used in tests."""
+    global _registry
+    _registry = None

headroom/learn/scanner.py

@@ -1,793 +1,26 @@
-"""Conversation scanners — read tool call logs from different agent systems.
+"""Conversation scanners — backwards-compatible re-exports.
 
-Scanners normalize conversation data into ToolCall sequences that analyzers
-can process regardless of the source system.
+Concrete scanner implementations have moved to headroom.learn.plugins.*.
+This module re-exports them so existing imports continue to work:
+
+    from headroom.learn.scanner import ClaudeCodeScanner  # still works
+    from headroom.learn.scanner import is_error_content    # still works
 """
 
 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,
-    SessionEvent,
-    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. Fall back to a
-            # simple slash replacement for display if we can't recover the real
-            # on-disk path from the filesystem.
-            project_path = _decode_project_path(entry.name)
-            if project_path is None:
-                # Fallback: detect Windows drive letter pattern (e.g., -C-MQ2-macros)
-                fallback_parts = entry.name[1:].split("-")
-                if len(fallback_parts[0]) == 1 and fallback_parts[0].isalpha():
-                    drive = fallback_parts[0].upper()
-                    project_path = Path(f"{drive}:\\" + "\\".join(fallback_parts[1:]))
-                else:
-                    project_path = Path("/" + entry.name[1:].replace("-", "/"))
-
-            # 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] = []
-        events: list[SessionEvent] = []
-        total_input_tokens = 0
-        total_output_tokens = 0
-        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", "")
-                    ts = d.get("timestamp", None)
-
-                    if line_type == "assistant":
-                        self._extract_tool_uses(d, tool_uses)
-                        # Extract token usage
-                        usage = d.get("message", {}).get("usage", {})
-                        total_input_tokens += usage.get("input_tokens", 0)
-                        total_input_tokens += usage.get("cache_read_input_tokens", 0)
-                        total_input_tokens += usage.get("cache_creation_input_tokens", 0)
-                        total_output_tokens += usage.get("output_tokens", 0)
-                    elif line_type == "user":
-                        self._extract_tool_results(d, tool_uses, tool_calls, events, msg_index, ts)
-                        self._extract_user_events(d, events, msg_index, ts)
-
-        except (OSError, UnicodeDecodeError) as e:
-            logger.debug("Failed to read %s: %s", jsonl_path, e)
-            return None
-
-        # Also wrap tool_calls as events for unified access
-        for tc in tool_calls:
-            if not any(e.type == "tool_call" and e.tool_call is tc for e in events):
-                events.append(SessionEvent(type="tool_call", msg_index=tc.msg_index, tool_call=tc))
-        events.sort(key=lambda e: e.msg_index)
-
-        return SessionData(
-            session_id=session_id,
-            tool_calls=tool_calls,
-            events=events,
-            total_input_tokens=total_input_tokens,
-            total_output_tokens=total_output_tokens,
-        )
-
-    def _extract_tool_uses(self, d: dict, tool_uses: dict[str, tuple[str, dict]]) -> None:
-        """Extract tool_use blocks from an assistant message."""
-        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],
-        events: list[SessionEvent],
-        msg_index: int,
-        timestamp: str | None = None,
-    ) -> None:
-        """Extract tool_result blocks from a user message and match to tool_uses."""
-        msg = d.get("message", {})
-        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
-
-            tc = ToolCall(
-                name=name,
-                tool_call_id=tc_id,
-                input_data=inp,
-                output=result_content,
-                is_error=is_err,
-                error_category=error_cat,
-                msg_index=msg_index,
-                output_bytes=len(result_content.encode("utf-8")),
-            )
-            tool_calls.append(tc)
-            events.append(
-                SessionEvent(
-                    type="tool_call", msg_index=msg_index, timestamp=timestamp, tool_call=tc
-                )
-            )
-
-            # Extract subagent summary from toolUseResult metadata
-            if name in ("Agent", "agent"):
-                tool_result_meta = d.get("toolUseResult", {})
-                if isinstance(tool_result_meta, dict):
-                    events.append(
-                        SessionEvent(
-                            type="agent_summary",
-                            msg_index=msg_index,
-                            timestamp=timestamp,
-                            agent_id=tool_result_meta.get("agentId", ""),
-                            agent_tool_count=tool_result_meta.get("totalToolUseCount", 0),
-                            agent_tokens=tool_result_meta.get("totalTokens", 0),
-                            agent_duration_ms=tool_result_meta.get("totalDurationMs", 0),
-                            agent_prompt=tool_result_meta.get("prompt", "")[:200],
-                        )
-                    )
-
-    def _extract_user_events(
-        self,
-        d: dict,
-        events: list[SessionEvent],
-        msg_index: int,
-        timestamp: str | None = None,
-    ) -> None:
-        """Extract user text messages and interruptions from a user line."""
-        msg = d.get("message", {})
-        content = msg.get("content", "")
-
-        # Human text messages have content as a string, not a list
-        if isinstance(content, str) and content.strip():
-            events.append(
-                SessionEvent(
-                    type="user_message",
-                    msg_index=msg_index,
-                    timestamp=timestamp,
-                    text=content[:500],
-                )
-            )
-            return
-
-        # Check for interruptions in list-format content
-        if isinstance(content, list):
-            for block in content:
-                if not isinstance(block, dict):
-                    continue
-                if block.get("type") == "text":
-                    text = block.get("text", "")
-                    if "[Request interrupted by user" in text:
-                        events.append(
-                            SessionEvent(
-                                type="interruption",
-                                msg_index=msg_index,
-                                timestamp=timestamp,
-                                text=text[:200],
-                            )
-                        )
-
-
-def _decode_project_path(escaped_name: str) -> Path | None:
-    """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
-
-    parts = escaped_name[1:].split("-")
-    if len(parts) < 2:
-        return None
-
-    # Windows drive letter detection: -C-Users-foo or -D-MQ2-macros
-    # First part is a single letter → treat as drive letter (C:\...)
-    if len(parts[0]) == 1 and parts[0].isalpha():
-        drive = parts[0].upper()
-        win_path = Path(f"{drive}:\\" + "\\".join(parts[1:]))
-        if win_path.exists():
-            return win_path
-        # Try greedy decode for Windows paths with hyphens in dir names
-        win_base = Path(f"{drive}:\\{parts[1]}") if len(parts) > 1 else win_path
-        if win_base.exists() and len(parts) > 2:
-            result = _greedy_path_decode(win_base, parts[2:])
-            if result:
-                return result
-
-    # Unix: simple approach — replace all - with / and check if path exists
-    simple = Path("/" + escaped_name[1:].replace("-", "/"))
-    if simple.exists():
-        return simple
-
-    # Try common Unix patterns: /Users/username/...
-    if len(parts) < 3:
-        return None
-
-    # Build path greedily: try joining with / and check existence
-    if parts[0] == "Users" and len(parts) > 2:
-        base = Path(f"/{parts[0]}/{parts[1]}")
-        remaining = parts[2:]
-        return _greedy_path_decode(base, remaining)
-
-    # Try /home/username/... (Linux)
-    if parts[0] == "home" and len(parts) > 2:
-        base = Path(f"/{parts[0]}/{parts[1]}")
-        remaining = parts[2:]
-        return _greedy_path_decode(base, remaining)
+# Shared helpers (moved to _shared.py)
+from ._shared import classify_error, is_error_content  # noqa: F401
 
-    return None
+# ConversationScanner ABC (canonical home is base.py)
+from .base import ConversationScanner  # noqa: F401
 
-
-def _greedy_path_decode(base: Path, parts: list[str]) -> Path | None:
-    """Greedily decode remaining path parts using real child directories.
-
-    Claude's project directory escaping is ambiguous: both ``/`` and literal
-    punctuation such as ``-`` or ``.`` may show up as ``-`` in the escaped
-    directory name. Instead of guessing every separator combination, walk the
-    actual filesystem and match each child directory against the remaining
-    escaped tokens.
-    """
-    if not parts:
-        return base if base.exists() else None
-
-    if not base.exists() or not base.is_dir():
-        return None
-
-    try:
-        children = sorted(child for child in base.iterdir() if child.is_dir())
-    except OSError:
-        return None
-
-    for child in children:
-        for tokenization in _component_tokenizations(child.name):
-            n_tokens = len(tokenization)
-            if parts[:n_tokens] != tokenization:
-                continue
-
-            result = _greedy_path_decode(child, parts[n_tokens:])
-            if result:
-                return result
-
-    return None
-
-
-def _component_tokenizations(component: str) -> list[list[str]]:
-    """Return possible escaped token sequences for a real path component."""
-    tokenizations: list[list[str]] = []
-    seen: set[tuple[str, ...]] = set()
-
-    def add(tokens: list[str]) -> None:
-        key = tuple(tokens)
-        if tokens and key not in seen:
-            seen.add(key)
-            tokenizations.append(tokens)
-
-    add([component])
-
-    for separator in ("-", ".", None):
-        if separator is None:
-            tokens = [token for token in re.split(r"[-.]", component) if token]
-        else:
-            tokens = [token for token in component.split(separator) if token]
-        add(tokens)
-
-    # Claude's flattened encoding can turn a leading "." in hidden directory
-    # names into an empty token followed by the remaining component tokens.
-    if component.startswith(".") and len(component) > 1:
-        hidden_component = component[1:]
-        add(["", hidden_component])
-        for separator in ("-", ".", None):
-            if separator is None:
-                tokens = [token for token in re.split(r"[-.]", hidden_component) if token]
-            else:
-                tokens = [token for token in hidden_component.split(separator) if token]
-            add(["", *tokens])
-
-    return tokenizations
-
-
-# =============================================================================
-# Codex Scanner (OpenAI Codex CLI)
-# =============================================================================
-
-
-class CodexScanner(ConversationScanner):
-    """Reads OpenAI Codex CLI session logs from ~/.codex/sessions/.
-
-    Codex stores sessions as JSON files with:
-    - session.id, session.timestamp, session.instructions
-    - items[]: array of message/function_call/function_call_output/reasoning objects
-
-    function_call items have: name, call_id, arguments (JSON string)
-    function_call_output items have: call_id, output (string or JSON string)
-    """
-
-    def __init__(self, codex_dir: Path | None = None):
-        self.codex_dir = codex_dir or Path.home() / ".codex"
-        self.sessions_dir = self.codex_dir / "sessions"
-
-    def _iter_session_files(self, root: Path | None = None) -> list[Path]:
-        """Return all known Codex session files, including nested rollouts."""
-        search_root = root or self.sessions_dir
-        session_files = list(search_root.rglob("*.json")) + list(search_root.rglob("*.jsonl"))
-        return sorted(path for path in session_files if path.is_file())
-
-    def discover_projects(self) -> list[ProjectInfo]:
-        """Codex doesn't organize by project — return a single 'codex' project.
-
-        Codex sessions aren't scoped to projects. We treat all sessions as one
-        project and use the cwd from session data (if available) to group later.
-        """
-        if not self.sessions_dir.exists():
-            return []
-
-        session_files = self._iter_session_files()
-        if not session_files:
-            return []
-
-        # Check for global AGENTS.md
-        agents_md = self.codex_dir / "AGENTS.md"
-        instructions_md = self.codex_dir / "instructions.md"
-
-        return [
-            ProjectInfo(
-                name="codex",
-                project_path=Path.cwd(),  # Codex doesn't track project paths
-                data_path=self.sessions_dir,
-                context_file=agents_md if agents_md.exists() else None,
-                memory_file=instructions_md if instructions_md.exists() else None,
-            )
-        ]
-
-    def scan_project(self, project: ProjectInfo) -> list[SessionData]:
-        """Scan all Codex session JSON files."""
-        sessions = []
-        for json_path in self._iter_session_files(project.data_path):
-            session = self._scan_session(json_path)
-            if session and session.tool_calls:
-                sessions.append(session)
-        return sessions
-
-    def _scan_session(self, json_path: Path) -> SessionData | None:
-        """Parse a single Codex session file."""
-        if json_path.suffix == ".jsonl":
-            return self._scan_jsonl_session(json_path)
-        return self._scan_json_session(json_path)
-
-    def _scan_json_session(self, json_path: Path) -> SessionData | None:
-        """Parse a single Codex session file."""
-        try:
-            with open(json_path) as f:
-                data = json.load(f)
-        except (OSError, json.JSONDecodeError) as e:
-            logger.debug("Failed to read Codex session %s: %s", json_path, e)
-            return None
-
-        session_info = data.get("session", {})
-        session_id = session_info.get("id", json_path.stem)
-        items = data.get("items", [])
-
-        if not items:
-            return None
-
-        # Build call_id → (name, input) map from function_call items
-        func_calls: dict[str, tuple[str, dict]] = {}
-        tool_calls: list[ToolCall] = []
-        msg_index = 0
-
-        for item in items:
-            msg_index += 1
-            item_type = item.get("type", "")
-
-            if item_type == "function_call":
-                call_id = item.get("call_id", "")
-                name = item.get("name", "")
-                # Parse arguments (JSON string or list)
-                raw_args = item.get("arguments", "")
-                if isinstance(raw_args, str):
-                    try:
-                        parsed = json.loads(raw_args)
-                    except (json.JSONDecodeError, TypeError):
-                        parsed = {"raw": raw_args}
-                elif isinstance(raw_args, dict):
-                    parsed = raw_args
-                else:
-                    parsed = {"raw": str(raw_args)}
-
-                # Codex uses "shell" as the tool name with command in args
-                # Normalize: extract command for consistency with other scanners
-                if name == "shell" and "command" in parsed:
-                    cmd = parsed["command"]
-                    if isinstance(cmd, list):
-                        # Codex passes commands as ["bash", "-lc", "actual command"]
-                        parsed["command"] = cmd[-1] if cmd else ""
-                    name = "Bash"  # Normalize to match Claude Code tool names
-
-                if call_id:
-                    func_calls[call_id] = (name, parsed)
-
-            elif item_type == "function_call_output":
-                call_id = item.get("call_id", "")
-                output_raw = item.get("output", "")
-
-                if call_id not in func_calls:
-                    continue
-
-                name, inp = func_calls[call_id]
-
-                # Output may be JSON string with "output" field
-                if isinstance(output_raw, str):
-                    try:
-                        parsed_out = json.loads(output_raw)
-                        if isinstance(parsed_out, dict) and "output" in parsed_out:
-                            result_content = str(parsed_out["output"])
-                        else:
-                            result_content = output_raw
-                    except (json.JSONDecodeError, TypeError):
-                        result_content = output_raw
-                else:
-                    result_content = str(output_raw)
-
-                is_err = is_error_content(result_content)
-                error_cat = classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
-
-                tool_calls.append(
-                    ToolCall(
-                        name=name,
-                        tool_call_id=call_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")),
-                    )
-                )
-
-        return SessionData(session_id=session_id, tool_calls=tool_calls)
-
-    def _scan_jsonl_session(self, jsonl_path: Path) -> SessionData | None:
-        """Parse a modern Codex rollout session stored as JSONL."""
-        session_id = jsonl_path.stem
-        func_calls: dict[str, tuple[str, dict]] = {}
-        tool_calls: list[ToolCall] = []
-        msg_index = 0
-
-        try:
-            with open(jsonl_path) as f:
-                for line in f:
-                    try:
-                        entry = json.loads(line)
-                    except json.JSONDecodeError:
-                        continue
-
-                    if entry.get("type") == "session_meta":
-                        payload = entry.get("payload", {})
-                        if isinstance(payload, dict):
-                            session_id = payload.get("id", session_id)
-                        continue
-
-                    if entry.get("type") != "response_item":
-                        continue
-
-                    payload = entry.get("payload", {})
-                    if not isinstance(payload, dict):
-                        continue
-
-                    msg_index += 1
-                    item_type = payload.get("type", "")
-
-                    if item_type in ("function_call", "custom_tool_call"):
-                        call_id = payload.get("call_id", "")
-                        name = payload.get("name", "")
-                        parsed = self._parse_codex_arguments(payload)
-                        name, parsed = self._normalize_codex_tool(name, parsed)
-                        if call_id and name:
-                            func_calls[call_id] = (name, parsed)
-                        continue
-
-                    if item_type not in ("function_call_output", "custom_tool_call_output"):
-                        continue
-
-                    call_id = payload.get("call_id", "")
-                    if call_id not in func_calls:
-                        continue
-
-                    name, inp = func_calls[call_id]
-                    result_content = self._parse_codex_output(payload.get("output", ""))
-                    is_err = is_error_content(result_content)
-                    error_cat = classify_error(result_content) if is_err else ErrorCategory.UNKNOWN
-
-                    tool_calls.append(
-                        ToolCall(
-                            name=name,
-                            tool_call_id=call_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")),
-                        )
-                    )
-
-        except OSError as e:
-            logger.debug("Failed to read Codex session %s: %s", jsonl_path, e)
-            return None
-
-        if not tool_calls:
-            return None
-
-        return SessionData(session_id=session_id, tool_calls=tool_calls)
-
-    def _parse_codex_arguments(self, payload: dict) -> dict:
-        """Parse arguments for either legacy or rollout Codex tool calls."""
-        raw_args = payload.get("arguments", payload.get("input", ""))
-        if isinstance(raw_args, str):
-            try:
-                parsed = json.loads(raw_args)
-                return parsed if isinstance(parsed, dict) else {"raw": raw_args}
-            except (json.JSONDecodeError, TypeError):
-                return {"raw": raw_args}
-        if isinstance(raw_args, dict):
-            return raw_args
-        return {"raw": str(raw_args)}
-
-    def _normalize_codex_tool(self, name: str, parsed: dict) -> tuple[str, dict]:
-        """Normalize modern Codex tool names to the cross-agent schema."""
-        if name == "shell" and "command" in parsed:
-            cmd = parsed["command"]
-            if isinstance(cmd, list):
-                parsed["command"] = cmd[-1] if cmd else ""
-            return "Bash", parsed
-
-        if name == "exec_command" and "cmd" in parsed:
-            parsed = dict(parsed)
-            parsed["command"] = parsed.get("cmd", "")
-            return "Bash", parsed
-
-        return name, parsed
-
-    def _parse_codex_output(self, output_raw: object) -> str:
-        """Parse tool output from Codex rollout records."""
-        if isinstance(output_raw, str):
-            try:
-                parsed_out = json.loads(output_raw)
-            except (json.JSONDecodeError, TypeError):
-                return output_raw
-
-            if isinstance(parsed_out, dict):
-                if "output" in parsed_out:
-                    return str(parsed_out["output"])
-                return json.dumps(parsed_out)
-            return output_raw
-
-        return str(output_raw)
+# Concrete scanners (moved to plugins/*, aliased to old names)
+from .plugins.claude import ClaudeCodePlugin as ClaudeCodeScanner  # noqa: F401
+from .plugins.claude import (  # noqa: F401
+    _component_tokenizations,
+    _decode_project_path,
+    _greedy_path_decode,
+)
+from .plugins.codex import CodexPlugin as CodexScanner  # noqa: F401
+from .plugins.gemini import GeminiPlugin as GeminiScanner  # noqa: F401

headroom/memory/traffic_learner.py

@@ -149,6 +149,7 @@ class TrafficLearner:
         self,
         backend: LocalBackend | None = None,
         user_id: str = "default",
+        agent_type: str = "unknown",
         max_history: int = 20,
         dedup_window: int = 100,
         min_evidence: int = 2,
@@ -159,12 +160,15 @@ class TrafficLearner:
             backend: Memory backend to save patterns to. If None, patterns
                 are accumulated but not persisted until a backend is set.
             user_id: Default user ID for saved memories.
+            agent_type: Which coding agent is being wrapped (claude, codex, gemini, etc.).
+                Used to determine the correct output file for flushing patterns.
             max_history: Number of recent tool calls to keep for pattern matching.
             dedup_window: Number of recent pattern hashes to track for dedup.
             min_evidence: Minimum times a pattern must be seen before saving.
         """
         self._backend = backend
         self._user_id = user_id
+        self.agent_type = agent_type
         self._max_history = max_history
         self._min_evidence = min_evidence
 
@@ -186,6 +190,7 @@ class TrafficLearner:
         # Background save queue
         self._save_queue: asyncio.Queue[ExtractedPattern] = asyncio.Queue(maxsize=100)
         self._save_task: asyncio.Task[None] | None = None
+        self._stopping = False
 
     # =========================================================================
     # Public API
@@ -201,14 +206,111 @@ class TrafficLearner:
             self._save_task = asyncio.create_task(self._save_worker())
 
     async def stop(self) -> None:
-        """Stop the background save worker."""
+        """Stop the background save worker, draining the queue first."""
+        # Drain any remaining patterns in the queue before cancelling
         if self._save_task and not self._save_task.done():
+            # Signal the worker to stop by putting a sentinel
+            self._stopping = True
             self._save_task.cancel()
             try:
                 await self._save_task
             except asyncio.CancelledError:
                 pass
 
+        # Drain any patterns left in the queue (worker may have been cancelled mid-flight)
+        while not self._save_queue.empty():
+            try:
+                pattern = self._save_queue.get_nowait()
+                if self._backend is not None:
+                    await self._backend.save_memory(
+                        content=pattern.content,
+                        user_id=self._user_id,
+                        metadata={
+                            "source": "traffic_learner",
+                            "category": pattern.category.value,
+                            "evidence_count": pattern.evidence_count,
+                            **pattern.metadata,
+                        },
+                    )
+                    self._patterns_saved += 1
+            except Exception:
+                break
+
+        # Flush learned patterns to the agent-native .md file
+        await self.flush_to_file()
+
+    async def flush_to_file(self) -> None:
+        """Flush accumulated patterns to the agent-native context file.
+
+        Uses the learn plugin registry to find the correct writer for the
+        current agent_type (e.g., MEMORY.md for Claude, AGENTS.md for Codex).
+        """
+        patterns = self.get_learned_patterns()
+        if not patterns or self.agent_type == "unknown":
+            return
+
+        try:
+            from headroom.learn.models import ProjectInfo, Recommendation, RecommendationTarget
+            from headroom.learn.registry import get_plugin
+
+            plugin = get_plugin(self.agent_type)
+            writer = plugin.create_writer()
+
+            # Convert patterns to Recommendations
+            recommendations = []
+            for p in patterns:
+                recommendations.append(
+                    Recommendation(
+                        target=RecommendationTarget.CONTEXT_FILE,
+                        section="Learned Patterns (Live Traffic)",
+                        content=f"- {p.content}",
+                        confidence=p.importance,
+                        evidence_count=p.evidence_count,
+                    )
+                )
+
+            if not recommendations:
+                return
+
+            # Use cwd as project path — the proxy runs in the project directory
+            from pathlib import Path
+
+            project = ProjectInfo(
+                name=Path.cwd().name,
+                project_path=Path.cwd(),
+                data_path=Path.cwd(),
+            )
+
+            result = writer.write(recommendations, project, dry_run=False)
+            if result.files_written:
+                logger.info(
+                    "Traffic learner flushed %d patterns to %s",
+                    len(recommendations),
+                    ", ".join(str(f) for f in result.files_written),
+                )
+        except KeyError:
+            logger.debug("No learn plugin for agent_type=%s, skipping file flush", self.agent_type)
+        except Exception as e:
+            logger.warning("Traffic learner flush_to_file failed: %s", e)
+
+    def get_learned_patterns(self) -> list[ExtractedPattern]:
+        """Return all patterns that have been saved or met the evidence threshold.
+
+        Includes patterns still in the accumulator that haven't hit min_evidence
+        but have been seen at least once (for end-of-session flush).
+        """
+        patterns: list[ExtractedPattern] = []
+
+        # Patterns that met the threshold and were queued
+        # (already saved to DB, but also want them in the .md file)
+        # We track what was saved via _saved_hashes, but don't keep the content.
+        # So we collect from the accumulator — patterns still accumulating.
+        for pattern, count in self._pattern_counts.values():
+            if count >= 1:  # At shutdown, flush even single-evidence patterns
+                patterns.append(pattern)
+
+        return patterns
+
     async def on_tool_result(
         self,
         tool_name: str,

headroom/proxy/models.py

@@ -176,6 +176,7 @@ class ProxyConfig:
     memory_db_path: str = "headroom_memory.db"
     memory_inject_tools: bool = True
     traffic_learning_enabled: bool = False
+    traffic_learning_agent_type: str = "unknown"  # Which agent is being wrapped
     memory_use_native_tool: bool = False
     memory_inject_context: bool = True
     memory_top_k: int = 10

headroom/proxy/server.py

@@ -447,11 +447,13 @@ class HeadroomProxy(
         # Traffic Learner (live pattern extraction from proxy traffic)
         # Only activates with --learn flag; requires --memory for backend
         self.traffic_learner: TrafficLearner | None = None
+        self.traffic_learning_agent_type: str = config.traffic_learning_agent_type
         if config.traffic_learning_enabled:
             from headroom.memory.traffic_learner import TrafficLearner
 
             self.traffic_learner = TrafficLearner(
                 user_id=os.environ.get("HEADROOM_USER_ID", os.environ.get("USER", "default")),
+                agent_type=config.traffic_learning_agent_type,
             )
 
     def _get_compression_cache(self, session_id: str) -> CompressionCache:

tests/test_learn/test_gemini_scanner.py

@@ -0,0 +1,531 @@
+"""Unit tests for GeminiScanner — Google Gemini CLI session parsing.
+
+Tests use synthetic session data in tmp directories, no real Gemini data needed.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from headroom.learn.models import ProjectInfo, Recommendation, RecommendationTarget
+from headroom.learn.scanner import GeminiScanner
+from headroom.learn.writer import GeminiWriter
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+
+def _make_gemini_session(
+    messages: list[dict],
+    session_id: str = "session-2026-04-09T10-00-abc123",
+) -> dict:
+    """Wrap messages in a Gemini session JSON structure."""
+    return {
+        "id": session_id,
+        "messages": messages,
+    }
+
+
+def _write_json_session(chats_dir: Path, data: dict, name: str = "session-test.json") -> Path:
+    """Write a session JSON file to a chats directory."""
+    path = chats_dir / name
+    path.write_text(json.dumps(data))
+    return path
+
+
+def _write_jsonl_session(
+    chats_dir: Path, records: list[dict], name: str = "session-test.jsonl"
+) -> Path:
+    """Write a session JSONL file to a chats directory."""
+    path = chats_dir / name
+    path.write_text("\n".join(json.dumps(r) for r in records))
+    return path
+
+
+def _setup_gemini_dir(tmp_path: Path) -> tuple[Path, Path]:
+    """Create ~/.gemini/tmp/<project>/chats/ directory structure."""
+    gemini_dir = tmp_path / ".gemini"
+    project_dir = gemini_dir / "tmp" / "abc123"
+    chats_dir = project_dir / "chats"
+    chats_dir.mkdir(parents=True)
+    return gemini_dir, chats_dir
+
+
+# =============================================================================
+# Project Discovery
+# =============================================================================
+
+
+class TestProjectDiscovery:
+    def test_no_gemini_dir(self, tmp_path):
+        scanner = GeminiScanner(gemini_dir=tmp_path / ".gemini")
+        assert scanner.discover_projects() == []
+
+    def test_empty_tmp_dir(self, tmp_path):
+        (tmp_path / ".gemini" / "tmp").mkdir(parents=True)
+        scanner = GeminiScanner(gemini_dir=tmp_path / ".gemini")
+        assert scanner.discover_projects() == []
+
+    def test_no_session_files(self, tmp_path):
+        chats_dir = tmp_path / ".gemini" / "tmp" / "proj1" / "chats"
+        chats_dir.mkdir(parents=True)
+        scanner = GeminiScanner(gemini_dir=tmp_path / ".gemini")
+        assert scanner.discover_projects() == []
+
+    def test_discovers_project_with_sessions(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {"role": "user", "parts": [{"text": "hello"}]},
+                {
+                    "role": "model",
+                    "parts": [
+                        {"functionCall": {"name": "read_file", "args": {"path": "/tmp/test.py"}}},
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "read_file",
+                                "response": {"output": "print('hello')"},
+                            }
+                        },
+                    ],
+                },
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        assert len(projects) == 1
+        assert projects[0].data_path == chats_dir
+
+
+# =============================================================================
+# JSON Session Parsing
+# =============================================================================
+
+
+class TestJsonSessionParsing:
+    def test_basic_tool_call(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {"role": "user", "parts": [{"text": "read the config file"}]},
+                {
+                    "role": "model",
+                    "parts": [
+                        {
+                            "functionCall": {
+                                "name": "read_file",
+                                "args": {"path": "/app/config.yaml"},
+                            }
+                        },
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "read_file",
+                                "response": {"output": "port: 8080\nhost: localhost"},
+                            }
+                        },
+                    ],
+                },
+                {"role": "model", "parts": [{"text": "The config has port 8080."}]},
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert len(sessions) == 1
+        assert len(sessions[0].tool_calls) == 1
+        tc = sessions[0].tool_calls[0]
+        assert tc.name == "Read"  # Normalized from read_file
+        assert tc.output == "port: 8080\nhost: localhost"
+        assert not tc.is_error
+
+    def test_multiple_tool_calls(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {"role": "user", "parts": [{"text": "find and read test files"}]},
+                {
+                    "role": "model",
+                    "parts": [
+                        {
+                            "functionCall": {
+                                "name": "search_files",
+                                "args": {"pattern": "test_*.py"},
+                            }
+                        },
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "search_files",
+                                "response": {"output": "tests/test_main.py\ntests/test_utils.py"},
+                            }
+                        },
+                    ],
+                },
+                {
+                    "role": "model",
+                    "parts": [
+                        {
+                            "functionCall": {
+                                "name": "read_file",
+                                "args": {"path": "tests/test_main.py"},
+                            }
+                        },
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "read_file",
+                                "response": {"output": "def test_main(): pass"},
+                            }
+                        },
+                    ],
+                },
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert len(sessions[0].tool_calls) == 2
+        assert sessions[0].tool_calls[0].name == "Glob"  # search_files → Glob
+        assert sessions[0].tool_calls[1].name == "Read"  # read_file → Read
+
+    def test_error_detection(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {
+                    "role": "model",
+                    "parts": [
+                        {"functionCall": {"name": "read_file", "args": {"path": "/missing.txt"}}},
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "read_file",
+                                "response": {
+                                    "output": "FileNotFoundError: No such file or directory: '/missing.txt'"
+                                },
+                            }
+                        },
+                    ],
+                },
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        tc = sessions[0].tool_calls[0]
+        assert tc.is_error
+        assert tc.error_category.value == "file_not_found"
+
+    def test_shell_command_normalized(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {
+                    "role": "model",
+                    "parts": [
+                        {
+                            "functionCall": {
+                                "name": "run_shell_command",
+                                "args": {"command": "ls -la"},
+                            }
+                        },
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "run_shell_command",
+                                "response": {
+                                    "output": "total 0\ndrwxr-xr-x 2 user user 64 Apr 9 10:00 ."
+                                },
+                            }
+                        },
+                    ],
+                },
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert sessions[0].tool_calls[0].name == "Bash"
+
+    def test_user_messages_extracted(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {"role": "user", "parts": [{"text": "What files are in this project?"}]},
+                {
+                    "role": "model",
+                    "parts": [
+                        {"functionCall": {"name": "search_files", "args": {"pattern": "*"}}},
+                    ],
+                },
+                {
+                    "role": "user",
+                    "parts": [
+                        {
+                            "functionResponse": {
+                                "name": "search_files",
+                                "response": {"output": "main.py"},
+                            }
+                        },
+                    ],
+                },
+                {"role": "user", "parts": [{"text": "Now run the tests"}]},
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        user_events = [e for e in sessions[0].events if e.type == "user_message"]
+        assert len(user_events) == 2
+        assert "What files" in user_events[0].text
+        assert "run the tests" in user_events[1].text
+
+    def test_array_format_messages(self, tmp_path):
+        """Sessions stored as bare array of messages (no wrapper object)."""
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        messages = [
+            {
+                "role": "model",
+                "parts": [
+                    {
+                        "functionCall": {
+                            "name": "write_file",
+                            "args": {"path": "test.py", "content": "pass"},
+                        }
+                    },
+                ],
+            },
+            {
+                "role": "user",
+                "parts": [
+                    {
+                        "functionResponse": {
+                            "name": "write_file",
+                            "response": {"output": "File written"},
+                        }
+                    },
+                ],
+            },
+        ]
+        _write_json_session(chats_dir, messages)  # Write array directly
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert len(sessions) == 1
+        assert sessions[0].tool_calls[0].name == "Write"
+
+    def test_no_tool_calls_returns_empty(self, tmp_path):
+        """Session with only text (no tool calls) produces no tool_calls."""
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        session = _make_gemini_session(
+            [
+                {"role": "user", "parts": [{"text": "What is Python?"}]},
+                {"role": "model", "parts": [{"text": "Python is a programming language."}]},
+            ]
+        )
+        _write_json_session(chats_dir, session)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        # No tool calls → session filtered out
+        assert len(sessions) == 0
+
+
+# =============================================================================
+# JSONL Session Parsing
+# =============================================================================
+
+
+class TestJsonlSessionParsing:
+    def test_basic_jsonl_session(self, tmp_path):
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        records = [
+            {"type": "session_metadata", "id": "ses-001"},
+            {"role": "user", "parts": [{"text": "list files"}]},
+            {
+                "role": "model",
+                "parts": [
+                    {"functionCall": {"name": "run_shell_command", "args": {"command": "ls"}}},
+                ],
+            },
+            {
+                "role": "user",
+                "parts": [
+                    {
+                        "functionResponse": {
+                            "name": "run_shell_command",
+                            "response": {"output": "main.py\ntest.py"},
+                        }
+                    },
+                ],
+            },
+        ]
+        _write_jsonl_session(chats_dir, records)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert len(sessions) == 1
+        assert sessions[0].tool_calls[0].name == "Bash"
+        assert "main.py" in sessions[0].tool_calls[0].output
+
+    def test_jsonl_type_field_roles(self, tmp_path):
+        """JSONL records where role is in the 'type' field (user/gemini)."""
+        gemini_dir, chats_dir = _setup_gemini_dir(tmp_path)
+        records = [
+            {
+                "type": "gemini",
+                "parts": [
+                    {"functionCall": {"name": "read_file", "args": {"path": "README.md"}}},
+                ],
+            },
+            {
+                "type": "user",
+                "parts": [
+                    {"functionResponse": {"name": "read_file", "response": {"output": "# Hello"}}},
+                ],
+            },
+        ]
+        _write_jsonl_session(chats_dir, records)
+
+        scanner = GeminiScanner(gemini_dir=gemini_dir)
+        projects = scanner.discover_projects()
+        sessions = scanner.scan_project(projects[0])
+
+        assert len(sessions) == 1
+        assert sessions[0].tool_calls[0].name == "Read"
+
+
+# =============================================================================
+# Tool Name Normalization
+# =============================================================================
+
+
+class TestToolNameNormalization:
+    def test_all_known_names(self):
+        from headroom.learn._shared import normalize_tool_name
+
+        assert normalize_tool_name("run_shell_command") == "Bash"
+        assert normalize_tool_name("shell") == "Bash"
+        assert normalize_tool_name("execute_command") == "Bash"
+        assert normalize_tool_name("read_file") == "Read"
+        assert normalize_tool_name("read_many_files") == "Read"
+        assert normalize_tool_name("write_file") == "Write"
+        assert normalize_tool_name("write_new_file") == "Write"
+        assert normalize_tool_name("create_file") == "Write"
+        assert normalize_tool_name("edit_file") == "Edit"
+        assert normalize_tool_name("replace_in_file") == "Edit"
+        assert normalize_tool_name("search_files") == "Glob"
+        assert normalize_tool_name("find_files") == "Glob"
+        assert normalize_tool_name("grep") == "Grep"
+        assert normalize_tool_name("search_text") == "Grep"
+        assert normalize_tool_name("list_directory") == "Glob"
+
+    def test_unknown_name_preserved(self):
+        from headroom.learn._shared import normalize_tool_name
+
+        assert normalize_tool_name("custom_tool") == "custom_tool"
+
+
+# =============================================================================
+# Writer Integration
+# =============================================================================
+
+
+class TestGeminiWriter:
+    def test_writes_to_gemini_md(self, tmp_path):
+        proj = ProjectInfo(name="gemini-test", project_path=tmp_path, data_path=tmp_path)
+        recs = [
+            Recommendation(
+                target=RecommendationTarget.CONTEXT_FILE,
+                section="Commands",
+                content="- Use `python -m pytest`",
+                confidence=0.9,
+                evidence_count=5,
+            ),
+        ]
+
+        writer = GeminiWriter()
+        result = writer.write(recs, proj, dry_run=False)
+
+        assert len(result.files_written) == 1
+        assert result.files_written[0].name == "GEMINI.md"
+        content = (tmp_path / "GEMINI.md").read_text()
+        assert "python -m pytest" in content
+
+    def test_empty_recs_no_write(self, tmp_path):
+        proj = ProjectInfo(name="clean", project_path=tmp_path, data_path=tmp_path)
+        writer = GeminiWriter()
+        result = writer.write([], proj, dry_run=False)
+        assert result.files_written == []
+        assert not (tmp_path / "GEMINI.md").exists()
+
+    def test_dry_run(self, tmp_path):
+        proj = ProjectInfo(name="test", project_path=tmp_path, data_path=tmp_path)
+        recs = [
+            Recommendation(
+                target=RecommendationTarget.CONTEXT_FILE,
+                section="Test",
+                content="- test",
+                confidence=0.8,
+                evidence_count=3,
+            ),
+        ]
+
+        writer = GeminiWriter()
+        result = writer.write(recs, proj, dry_run=True)
+
+        assert result.dry_run is True
+        assert len(result.files_written) == 1
+        # Dry run should NOT create the file
+        assert not (tmp_path / "GEMINI.md").exists()

tests/test_learn/test_registry.py

@@ -0,0 +1,127 @@
+"""Tests for the learn plugin registry."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from headroom.learn.base import LearnPlugin
+from headroom.learn.registry import (
+    auto_detect_plugins,
+    available_agent_names,
+    get_plugin,
+    get_registry,
+    reset_registry,
+)
+from headroom.learn.scanner import ConversationScanner
+
+
+@pytest.fixture(autouse=True)
+def _clean_registry():
+    """Reset registry before/after each test."""
+    reset_registry()
+    yield
+    reset_registry()
+
+
+class TestBuiltinDiscovery:
+    def test_discovers_three_builtin_plugins(self):
+        reg = get_registry()
+        assert "claude" in reg
+        assert "codex" in reg
+        assert "gemini" in reg
+        assert len(reg) >= 3
+
+    def test_all_plugins_are_learn_plugins(self):
+        for name, plugin in get_registry().items():
+            assert isinstance(plugin, LearnPlugin), f"{name} is not a LearnPlugin"
+
+    def test_all_plugins_are_conversation_scanners(self):
+        """Backwards compat: plugins must also be ConversationScanners."""
+        for name, plugin in get_registry().items():
+            assert isinstance(plugin, ConversationScanner), f"{name} is not a ConversationScanner"
+
+    def test_plugins_have_identity(self):
+        for name, plugin in get_registry().items():
+            assert plugin.name == name
+            assert plugin.display_name  # non-empty
+            assert plugin.description  # non-empty
+
+
+class TestGetPlugin:
+    def test_get_existing_plugin(self):
+        plugin = get_plugin("claude")
+        assert plugin.name == "claude"
+        assert plugin.display_name == "Claude Code"
+
+    def test_get_unknown_raises_keyerror(self):
+        with pytest.raises(KeyError, match="Unknown agent.*cursor"):
+            get_plugin("cursor")
+
+    def test_error_message_lists_available(self):
+        with pytest.raises(KeyError, match="claude"):
+            get_plugin("nonexistent")
+
+
+class TestAutoDetect:
+    def test_filters_to_detected_only(self):
+        """Only plugins where detect() returns True are included."""
+        detected = auto_detect_plugins()
+        for plugin in detected:
+            assert plugin.detect()
+
+    def test_returns_empty_when_nothing_detected(self):
+        """All plugins returning False → empty list."""
+        with (
+            patch.object(get_registry()["claude"], "detect", return_value=False),
+            patch.object(get_registry()["codex"], "detect", return_value=False),
+            patch.object(get_registry()["gemini"], "detect", return_value=False),
+        ):
+            assert auto_detect_plugins() == []
+
+
+class TestAvailableNames:
+    def test_returns_sorted_list(self):
+        names = available_agent_names()
+        assert names == sorted(names)
+        assert "claude" in names
+        assert "codex" in names
+        assert "gemini" in names
+
+
+class TestExternalPlugin:
+    def test_external_plugin_via_entry_point(self):
+        """Mock an external plugin registered via entry_points."""
+        mock_plugin = MagicMock(spec=LearnPlugin)
+        mock_plugin.name = "cursor"
+        mock_plugin.display_name = "Cursor"
+        mock_plugin.description = "Cursor IDE (~/.cursor/)"
+
+        mock_ep = MagicMock()
+        mock_ep.load.return_value = mock_plugin
+        mock_ep.name = "cursor"
+
+        with patch("importlib.metadata.entry_points", return_value=[mock_ep]):
+            reset_registry()
+            reg = get_registry()
+            assert "cursor" in reg
+            assert reg["cursor"].name == "cursor"
+
+
+class TestResetRegistry:
+    def test_reset_clears_cache(self):
+        reg1 = get_registry()
+        assert reg1 is get_registry()  # Same object (cached)
+        reset_registry()
+        reg2 = get_registry()
+        assert reg2 is not reg1  # New object (cache cleared)
+
+
+class TestPluginCreateWriter:
+    def test_all_plugins_create_valid_writers(self):
+        from headroom.learn.writer import ContextWriter
+
+        for name, plugin in get_registry().items():
+            writer = plugin.create_writer()
+            assert isinstance(writer, ContextWriter), f"{name} writer is not a ContextWriter"