fix(learn): scan subagent and workflow transcripts (#1045)

## Description

`headroom learn` only scanned top-level main Claude Code sessions
(`<project>/<uuid>.jsonl`). Nested subagent and workflow transcripts
under `<project>/<uuid>/subagents/**` were not opened, which hid a large
amount of tool-call failure and token-spend activity from failure mining
and downstream analysis.

This change makes the Claude scanner descend into nested transcripts by
default and tag each `SessionData` with its source. `--main-only`
restores the previous top-level-only scan scope.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- Updated `ClaudeCodePlugin.scan_project` to discover nested subagent
and workflow transcripts by default.
- Added source tagging for `main`, `subagent`, and `workflow` sessions.
- Added `--main-only` and `include_subagents` plumbing so callers can
opt back into top-level-only scanning.
- Added the `include_subagents` scanner parameter to Codex/Gemini as a
documented no-op because those scanners use flat session layouts.
- Added regression tests for nested discovery, source tagging, parallel
scanning, and CLI flag threading.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality

### Test Output

```text
Full learn + CLI suite:
# 186 passed, 2 skipped

GitHub Actions CI for this PR:
# build, build-wheel, lint, tests, e2e, CodeQL, and native wrapper checks passed
```

## Real Behavior Proof

- Environment: local Claude Code corpus with nested subagent/workflow
transcripts.
- Exact command / steps: Scanned the corpus with the previous
top-level-only behavior and then with nested transcript discovery
enabled.
- Observed result: The scanner saw 24 sessions before and 306 sessions
after descending into nested transcripts.
- Not tested: Codex/Gemini nested transcript discovery, because those
providers currently use flat session layouts and treat
`include_subagents` as a no-op.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

---------

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

headroom/cli/learn.py

@@ -98,6 +98,13 @@ Use 'auto' (default) to scan all detected agents."""
     help="Parallel workers for session scanning. "
     "Default: auto (min of CPU count, 8). Use 1 for serial.",
 )
+@click.option(
+    "--main-only",
+    is_flag=True,
+    default=False,
+    help="Only scan top-level main sessions, skipping nested subagent/workflow "
+    "transcripts (Claude Code). Default scans everything.",
+)
 def learn(
     project: Path | None,
     analyze_all: bool,
@@ -105,6 +112,7 @@ def learn(
     agent: str,
     model: str | None,
     workers: int | None,
+    main_only: bool,
 ) -> None:
     """Learn from past tool call failures to prevent future ones.
 
@@ -198,7 +206,9 @@ def learn(
             click.echo(f"Path: {proj.project_path}")
             click.echo(f"{'=' * 60}")
 
-            sessions = plugin.scan_project(proj, max_workers=max_workers)
+            sessions = plugin.scan_project(
+                proj, max_workers=max_workers, include_subagents=not main_only
+            )
             if not sessions:
                 click.echo("  No conversation data found.")
                 continue

headroom/learn/base.py

@@ -25,7 +25,9 @@ class ConversationScanner(ABC):
         ...
 
     @abstractmethod
-    def scan_project(self, project: ProjectInfo, max_workers: int = 1) -> list[SessionData]:
+    def scan_project(
+        self, project: ProjectInfo, max_workers: int = 1, include_subagents: bool = True
+    ) -> list[SessionData]:
         """Scan all sessions for a project, returning normalized tool calls."""
         ...
 
@@ -52,7 +54,12 @@ class LearnPlugin(ABC):
                 return Path("~/.myagent/sessions").expanduser().exists()
 
             def discover_projects(self) -> list[ProjectInfo]: ...
-            def scan_project(self, project: ProjectInfo) -> list[SessionData]: ...
+            def scan_project(
+                self,
+                project: ProjectInfo,
+                max_workers: int = 1,
+                include_subagents: bool = True,
+            ) -> list[SessionData]: ...
 
             def create_writer(self) -> ContextWriter:
                 from headroom.learn.writer import GeminiWriter
@@ -99,13 +106,18 @@ class LearnPlugin(ABC):
         ...
 
     @abstractmethod
-    def scan_project(self, project: ProjectInfo, max_workers: int = 1) -> list[SessionData]:
+    def scan_project(
+        self, project: ProjectInfo, max_workers: int = 1, include_subagents: bool = True
+    ) -> list[SessionData]:
         """Scan all sessions for a project, returning normalized data.
 
         Args:
             project: The project to scan.
             max_workers: Number of threads for parallel file scanning.
                          1 (default) = serial.  >1 = concurrent.
+            include_subagents: Also scan nested subagent/workflow transcripts
+                         where the agent system writes them (Claude Code).
+                         Ignored by agents without a nested transcript layout.
         """
         ...
 

headroom/learn/models.py

@@ -112,6 +112,7 @@ class SessionData:
     timestamp: datetime | None = None
     total_input_tokens: int = 0
     total_output_tokens: int = 0
+    source: str = "main"  # "main" | "subagent" | "workflow" — where this transcript came from
 
     @property
     def failure_count(self) -> int:

headroom/learn/plugins/claude.py

@@ -106,27 +106,57 @@ class ClaudeCodePlugin(LearnPlugin, ConversationScanner):
 
         return projects
 
-    def scan_project(self, project: ProjectInfo, max_workers: int = 1) -> list[SessionData]:
-        """Scan all conversation JSONL files for a project."""
-        jsonl_files = sorted(project.data_path.glob("*.jsonl"))
+    def scan_project(
+        self, project: ProjectInfo, max_workers: int = 1, include_subagents: bool = True
+    ) -> list[SessionData]:
+        """Scan all conversation JSONL files for a project.
+
+        Claude Code writes the main session at ``<project>/<uuid>.jsonl`` and
+        nests the transcripts it spawns under ``<project>/<uuid>/subagents/**``
+        (subagents) and ``.../subagents/workflows/**`` (workflow agents). Each
+        nested transcript is its own context window with its own token spend, so
+        by default we descend into them. Pass ``include_subagents=False`` to
+        restrict to top-level main sessions only.
+        """
+        data_path = project.data_path
+        if include_subagents:
+            jsonl_files = sorted(data_path.rglob("*.jsonl"))
+        else:
+            jsonl_files = sorted(data_path.glob("*.jsonl"))
         if not jsonl_files:
             return []
 
+        file_sources = [(f, self._classify_source(data_path, f)) for f in jsonl_files]
+
         if max_workers <= 1 or len(jsonl_files) <= 1:
-            return [s for f in jsonl_files if (s := self._scan_session(f)) and s.tool_calls]
+            return [
+                s
+                for f, src in file_sources
+                if (s := self._scan_session(f, source=src)) and s.tool_calls
+            ]
 
         from concurrent.futures import ThreadPoolExecutor, as_completed
 
         sessions: list[SessionData] = []
         with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = {executor.submit(self._scan_session, f): f for f in jsonl_files}
+            futures = {executor.submit(self._scan_session, f, src): f for f, src in file_sources}
             for future in as_completed(futures):
                 session = future.result()
                 if session and session.tool_calls:
                     sessions.append(session)
         return sessions
 
-    def _scan_session(self, jsonl_path: Path) -> SessionData | None:
+    @staticmethod
+    def _classify_source(data_path: Path, jsonl_path: Path) -> str:
+        """Tag a transcript as main / subagent / workflow from its path depth."""
+        parts = jsonl_path.relative_to(data_path).parts
+        if len(parts) == 1:
+            return "main"
+        if "workflows" in parts:
+            return "workflow"
+        return "subagent"
+
+    def _scan_session(self, jsonl_path: Path, source: str = "main") -> SessionData | None:
         """Scan a single JSONL conversation file."""
         session_id = jsonl_path.stem
         tool_uses: dict[str, tuple[str, dict]] = {}
@@ -174,6 +204,7 @@ class ClaudeCodePlugin(LearnPlugin, ConversationScanner):
             events=events,
             total_input_tokens=total_input_tokens,
             total_output_tokens=total_output_tokens,
+            source=source,
         )
 
     def _extract_tool_uses(self, d: dict, tool_uses: dict[str, tuple[str, dict]]) -> None:

headroom/learn/plugins/codex.py

@@ -91,8 +91,14 @@ class CodexPlugin(LearnPlugin, ConversationScanner):
             )
         ]
 
-    def scan_project(self, project: ProjectInfo, max_workers: int = 1) -> list[SessionData]:
-        """Scan all Codex session JSON files."""
+    def scan_project(
+        self, project: ProjectInfo, max_workers: int = 1, include_subagents: bool = True
+    ) -> list[SessionData]:
+        """Scan all Codex session JSON files.
+
+        ``include_subagents`` is accepted for a uniform plugin contract but is a
+        no-op: Codex stores sessions flat, with no nested transcript hierarchy.
+        """
         session_files = self._iter_session_files(project.data_path)
         if not session_files:
             return []

headroom/learn/plugins/gemini.py

@@ -106,8 +106,14 @@ class GeminiPlugin(LearnPlugin, ConversationScanner):
 
         return projects
 
-    def scan_project(self, project: ProjectInfo, max_workers: int = 1) -> list[SessionData]:
-        """Scan all Gemini session files for a project."""
+    def scan_project(
+        self, project: ProjectInfo, max_workers: int = 1, include_subagents: bool = True
+    ) -> list[SessionData]:
+        """Scan all Gemini session files for a project.
+
+        ``include_subagents`` is accepted for a uniform plugin contract but is a
+        no-op: Gemini stores sessions flat, with no nested transcript hierarchy.
+        """
         session_files = sorted(project.data_path.glob("session-*.json")) + sorted(
             project.data_path.glob("session-*.jsonl")
         )

tests/test_cli_learn.py

@@ -42,6 +42,7 @@ class FakePlugin:
         self._projects = projects
         self.writer = FakeWriter()
         self.scan_calls: list[tuple[object, int]] = []
+        self.last_include_subagents: bool | None = None
 
     def detect(self) -> bool:
         return True
@@ -52,8 +53,9 @@ class FakePlugin:
     def discover_projects(self) -> list[object]:
         return self._projects
 
-    def scan_project(self, project, max_workers: int = 1):  # noqa: ANN001, ANN201
+    def scan_project(self, project, max_workers: int = 1, include_subagents: bool = True):  # noqa: ANN001, ANN201
         self.scan_calls.append((project, max_workers))
+        self.last_include_subagents = include_subagents
         return [SimpleNamespace(events=["event"], tool_calls=[], failure_count=0)]
 
 
@@ -259,7 +261,7 @@ def test_learn_handles_empty_sessions_and_no_pattern_outputs(
     no_actions = SimpleNamespace(name="no-actions", project_path=tmp_path / "no-actions")
 
     class BranchingPlugin(FakePlugin):
-        def scan_project(self, project, max_workers: int = 1):  # noqa: ANN001, ANN201
+        def scan_project(self, project, max_workers: int = 1, include_subagents: bool = True):  # noqa: ANN001, ANN201
             self.scan_calls.append((project, max_workers))
             if project is no_sessions:
                 return []
@@ -297,3 +299,29 @@ def test_learn_handles_empty_sessions_and_no_pattern_outputs(
     assert "No conversation data found." in result.output
     assert "No failures or patterns found." in result.output
     assert "No actionable patterns found." in result.output
+
+
+def test_learn_main_only_flag_threads_to_scanner(
+    monkeypatch: pytest.MonkeyPatch, runner: CliRunner, tmp_path: Path
+) -> None:
+    project_path = tmp_path / "proj"
+    project_path.mkdir()
+    proj = SimpleNamespace(name="proj", project_path=project_path)
+    plugin = FakePlugin("codex", "Codex", [proj])
+
+    monkeypatch.setattr("headroom.learn.analyzer._detect_default_model", lambda: "gpt-4o")
+    monkeypatch.setattr("headroom.learn.registry.get_plugin", lambda name: plugin)
+    monkeypatch.setattr("headroom.learn.analyzer.SessionAnalyzer", FakeAnalyzer)
+
+    # Default: descend into subagent/workflow transcripts.
+    result = runner.invoke(main, ["learn", "--agent", "codex", "--all"], catch_exceptions=False)
+    assert result.exit_code == 0, result.output
+    assert plugin.last_include_subagents is True
+
+    # --main-only restricts to top-level main sessions.
+    plugin.last_include_subagents = None
+    result = runner.invoke(
+        main, ["learn", "--agent", "codex", "--all", "--main-only"], catch_exceptions=False
+    )
+    assert result.exit_code == 0, result.output
+    assert plugin.last_include_subagents is False

tests/test_learn/test_subagent_scanning.py

@@ -0,0 +1,81 @@
+"""The Claude scanner must descend into subagent and workflow transcripts.
+
+Claude Code writes a main session at ``<project>/<uuid>.jsonl`` and nests the
+transcripts it spawns under ``<project>/<uuid>/subagents/**`` (subagents) and
+``.../subagents/workflows/**`` (workflow agents). Each nested transcript is a
+separate context window with its own token spend and its own tool-call
+failures, so ``headroom learn`` must see them — not just the top-level session.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from headroom.learn.models import ProjectInfo
+from headroom.learn.plugins.claude import ClaudeCodePlugin
+
+
+def _write_session(path: Path, out: str = "x" * 400) -> None:
+    """Write a minimal Claude Code session: one tool_use paired with a result."""
+    lines = [
+        {
+            "type": "assistant",
+            "message": {
+                "usage": {"input_tokens": 100, "output_tokens": 10},
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "u1",
+                        "name": "Read",
+                        "input": {"file_path": "/a.py"},
+                    }
+                ],
+            },
+        },
+        {
+            "type": "user",
+            "message": {"content": [{"type": "tool_result", "tool_use_id": "u1", "content": out}]},
+        },
+    ]
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(json.dumps(line) for line in lines))
+
+
+def test_scan_project_discovers_subagent_and_workflow_transcripts(tmp_path: Path) -> None:
+    _write_session(tmp_path / "main-uuid.jsonl")
+    _write_session(tmp_path / "main-uuid" / "subagents" / "agent-1.jsonl")
+    _write_session(tmp_path / "main-uuid" / "subagents" / "workflows" / "wf_1" / "agent-2.jsonl")
+
+    plugin = ClaudeCodePlugin()
+    project = ProjectInfo(name="p", project_path=tmp_path, data_path=tmp_path)
+    sessions = plugin.scan_project(project, max_workers=1)
+
+    assert len(sessions) == 3
+    assert sorted(s.source for s in sessions) == ["main", "subagent", "workflow"]
+
+
+def test_main_only_restricts_to_top_level(tmp_path: Path) -> None:
+    _write_session(tmp_path / "main-uuid.jsonl")
+    _write_session(tmp_path / "main-uuid" / "subagents" / "agent-1.jsonl")
+
+    plugin = ClaudeCodePlugin()
+    project = ProjectInfo(name="p", project_path=tmp_path, data_path=tmp_path)
+    sessions = plugin.scan_project(project, max_workers=1, include_subagents=False)
+
+    assert len(sessions) == 1
+    assert sessions[0].source == "main"
+
+
+def test_subagents_found_in_parallel_scan(tmp_path: Path) -> None:
+    # Multiple files force the ThreadPool path; nested transcripts must still appear.
+    _write_session(tmp_path / "main-a.jsonl")
+    _write_session(tmp_path / "main-b.jsonl")
+    _write_session(tmp_path / "main-a" / "subagents" / "agent-1.jsonl")
+
+    plugin = ClaudeCodePlugin()
+    project = ProjectInfo(name="p", project_path=tmp_path, data_path=tmp_path)
+    sessions = plugin.scan_project(project, max_workers=4)
+
+    assert len(sessions) == 3
+    assert sum(1 for s in sessions if s.source == "subagent") == 1