Fix training data export pipeline and TB-5 cross-repo br calls

Training exporters searched only Claude Code session logs (~/.claude/projects/)
with *.jsonl glob. Pipeline agent sessions live at ~/.local/share/dev-loop/sessions/
as *.ndjson files with a different format (single JSON array per line).

Added shared helpers in training/__init__.py:
- _default_pipeline_sessions_dir(): returns pipeline sessions path
- _collect_session_files(): searches both dirs with both extensions
- _load_session_events(): handles jsonl (one obj/line) and ndjson (array) formats
- _is_external_user(): detects user events across both formats

Updated all three exporters to use the shared helpers. Persona exporter now
reads .meta.json sidecars for authoritative persona/success data (12 examples
exported, up from 0). Fixed TB-5 cascade br calls to use source/target repo
paths instead of _DEVLOOP_ROOT for cross-repo issue operations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: musicofhel

src/devloop/feedback/tb5_cascade.py

@@ -178,7 +178,7 @@ def _get_source_issue_details(issue_id: str, repo_path: str | None = None) -> di
         text=True,
         check=False,
         timeout=30,
-        cwd=_DEVLOOP_ROOT,
+        cwd=repo_path or _DEVLOOP_ROOT,
     )
     if result.returncode != 0:
         error_msg = result.stderr.strip() or f"br show failed with exit code {result.returncode}"
@@ -222,6 +222,7 @@ def _create_cascade_issue(
     labels = f"cascade,repo:{target_repo_name}"
 
     # Try with --parent first (works when source+target share a beads workspace)
+    cwd = repo_path or _DEVLOOP_ROOT
     cmd = [
         "br", "create", title,
         "--description", description,
@@ -235,7 +236,7 @@ def _create_cascade_issue(
         text=True,
         check=False,
         timeout=30,
-        cwd=_DEVLOOP_ROOT,
+        cwd=cwd,
     )
 
     # If --parent fails (cross-repo: parent issue not in target beads), retry without it
@@ -256,7 +257,7 @@ def _create_cascade_issue(
             text=True,
             check=False,
             timeout=30,
-            cwd=_DEVLOOP_ROOT,
+            cwd=cwd,
         )
 
     if result.returncode != 0:
@@ -303,7 +304,7 @@ def _report_cascade_outcome(
         text=True,
         check=False,
         timeout=30,
-        cwd=_DEVLOOP_ROOT,
+        cwd=repo_path or _DEVLOOP_ROOT,
     )
     if result.returncode != 0:
         logger.warning(

src/devloop/llmops/training/__init__.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import glob
 import json
 import os
 import shutil
@@ -20,6 +21,79 @@ def _default_sessions_dir() -> str:
     return os.path.expanduser(f"~/.claude/projects/-{mangled}/")
 
 
+def _default_pipeline_sessions_dir() -> str:
+    """Dev-loop pipeline agent sessions directory.
+
+    TB runs store agent sessions as .ndjson files with .meta.json sidecars
+    at ``~/.local/share/dev-loop/sessions/``.
+    """
+    return os.path.expanduser("~/.local/share/dev-loop/sessions/")
+
+
+def _collect_session_files(
+    sessions_dir: str | None = None,
+    max_files: int = 200,
+) -> list[str]:
+    """Collect session files from Claude Code logs AND pipeline sessions.
+
+    When *sessions_dir* is provided, searches that single directory for both
+    ``.jsonl`` and ``.ndjson`` files.  When ``None``, searches both the
+    default Claude Code sessions dir and the pipeline sessions dir.
+    """
+    if sessions_dir is not None:
+        dirs = [sessions_dir]
+    else:
+        dirs = [_default_sessions_dir(), _default_pipeline_sessions_dir()]
+
+    files: list[str] = []
+    for d in dirs:
+        if not os.path.isdir(d):
+            continue
+        files.extend(glob.glob(os.path.join(d, "*.jsonl")))
+        files.extend(glob.glob(os.path.join(d, "*.ndjson")))
+
+    return sorted(set(files))[:max_files]
+
+
+def _load_session_events(fpath: str) -> list[dict]:
+    """Load events from a ``.jsonl`` or ``.ndjson`` session file.
+
+    Handles two formats:
+    - ``.jsonl``: one JSON object per line (Claude Code conversation logs).
+    - ``.ndjson``: a single JSON array containing all events on one line
+      (pipeline agent sessions).
+    """
+    events: list[dict] = []
+    with open(fpath) as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                parsed = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if isinstance(parsed, list):
+                # ndjson: single array of event dicts
+                events.extend(e for e in parsed if isinstance(e, dict))
+            elif isinstance(parsed, dict):
+                events.append(parsed)
+    return events
+
+
+def _is_external_user(evt: dict) -> bool:
+    """Check if an event is from an external/human user.
+
+    Claude Code ``.jsonl`` events use ``userType: "external"``.
+    Pipeline ``.ndjson`` events use ``type: "user"`` without ``userType``.
+    """
+    if evt.get("userType") == "external":
+        return True
+    if evt.get("type") == "user" and "userType" not in evt:
+        return True
+    return False
+
+
 def safe_write_jsonl(
     output_path: str,
     examples: list[dict],

src/devloop/llmops/training/export_personas.py

@@ -16,6 +16,12 @@
 
 from opentelemetry import trace
 
+from devloop.llmops.training import (
+    _collect_session_files,
+    _is_external_user,
+    _load_session_events,
+)
+
 tracer = trace.get_tracer("llmops.training", "0.1.0")
 
 
@@ -131,7 +137,7 @@ def _extract_persona_data(events: list[dict]) -> dict | None:
                     break
 
         # Capture first substantial human message as task description
-        if not issue_description and evt.get("userType") == "external" and len(content) > 30:
+        if not issue_description and _is_external_user(evt) and len(content) > 30:
             issue_description = content[:2000]
 
         # Detect outcome
@@ -169,11 +175,6 @@ def export_personas(
     """
     from devloop.llmops.training import safe_write_jsonl
 
-    if sessions_dir is None:
-        from devloop.llmops.training import _default_sessions_dir
-
-        sessions_dir = _default_sessions_dir()
-
     if output_path is None:
         output_path = os.path.expanduser(
             "~/.local/share/dev-loop/llmops/training/persona_select.jsonl"
@@ -185,27 +186,66 @@ def export_personas(
     ) as span:
         Path(output_path).parent.mkdir(parents=True, exist_ok=True)
 
-        files = sorted(glob.glob(os.path.join(sessions_dir, "*.jsonl")))[:max_sessions]
+        files = _collect_session_files(sessions_dir, max_sessions)
         examples: list[dict] = []
 
         for fpath in files:
-            events = []
-            with open(fpath) as f:
-                for line in f:
-                    line = line.strip()
-                    if not line:
-                        continue
-                    try:
-                        events.append(json.loads(line))
-                    except json.JSONDecodeError:
-                        continue
+            events = _load_session_events(fpath)
+            stem = Path(fpath).stem
+
+            # For pipeline sessions, try the .meta.json sidecar first —
+            # it has authoritative persona/success data from orchestration.
+            meta_path = re.sub(r"\.(ndjson|jsonl)$", ".meta.json", fpath)
+            meta: dict | None = None
+            if os.path.isfile(meta_path):
+                try:
+                    with open(meta_path) as mf:
+                        meta = json.load(mf)
+                except (json.JSONDecodeError, OSError):
+                    meta = None
+
+            if meta and meta.get("persona"):
+                # Build example directly from structured metadata.
+                issue_desc = ""
+                for evt in events:
+                    msg = evt.get("message", {})
+                    content = msg.get("content", "")
+                    if isinstance(content, list):
+                        for block in content:
+                            if isinstance(block, dict) and block.get("type") == "text":
+                                content = block.get("text", "")
+                                break
+                    if isinstance(content, str) and _is_external_user(evt) and len(content) > 30:
+                        issue_desc = content[:2000]
+                        break
+
+                repo_type = _detect_repo_type(events)
+                example = {
+                    "inputs": {
+                        "issue_labels": "",
+                        "issue_description": issue_desc or meta.get("issue_id", ""),
+                        "repo_type": repo_type,
+                    },
+                    "outputs": {
+                        "persona_id": meta["persona"],
+                        "custom_guidelines": "",
+                        "task_succeeded": str(meta.get("success", False)),
+                    },
+                    "metadata": {
+                        "session_id": stem,
+                        "source": "persona_meta",
+                    },
+                }
+                examples.append(example)
+                continue
 
+            # Fall back to parsing persona from conversation text.
             example = _extract_persona_data(events)
             if example is None:
                 continue
 
             example["metadata"] = {
-                "session_id": os.path.basename(fpath).replace(".jsonl", ""),
+                "session_id": stem,
                 "source": "persona_session",
             }
             examples.append(example)
@@ -219,13 +259,9 @@ def export_personas(
 if __name__ == "__main__":
     import sys
 
-    from devloop.llmops.training import _default_sessions_dir
-
-    sessions_dir = _default_sessions_dir()
-    if not os.path.isdir(sessions_dir):
-        print(f"WARNING: Sessions dir not found: {sessions_dir}", file=sys.stderr)
-    elif not glob.glob(os.path.join(sessions_dir, "*.jsonl")):
-        print(f"WARNING: No .jsonl files in {sessions_dir}", file=sys.stderr)
+    files = _collect_session_files()
+    if not files:
+        print("WARNING: No session files found", file=sys.stderr)
 
     force = "--force" in sys.argv
     count = export_personas(force=force)

src/devloop/llmops/training/export_retries.py

@@ -16,6 +16,12 @@
 
 from opentelemetry import trace
 
+from devloop.llmops.training import (
+    _collect_session_files,
+    _is_external_user,
+    _load_session_events,
+)
+
 tracer = trace.get_tracer("llmops.training", "0.1.0")
 
 
@@ -70,7 +76,7 @@ def _extract_retry_data(events: list[dict]) -> list[dict]:
             continue
 
         # Capture original task from first human message
-        if not original_task and evt.get("userType") == "external":
+        if not original_task and _is_external_user(evt):
             if not _is_retry_prompt(content) and len(content) > 20:
                 original_task = content[:2000]
 
@@ -146,11 +152,6 @@ def export_retries(
     """
     from devloop.llmops.training import safe_write_jsonl
 
-    if sessions_dir is None:
-        from devloop.llmops.training import _default_sessions_dir
-
-        sessions_dir = _default_sessions_dir()
-
     if output_path is None:
         output_path = os.path.expanduser(
             "~/.local/share/dev-loop/llmops/training/retry_prompt.jsonl"
@@ -162,25 +163,14 @@ def export_retries(
     ) as span:
         Path(output_path).parent.mkdir(parents=True, exist_ok=True)
 
-        files = sorted(glob.glob(os.path.join(sessions_dir, "*.jsonl")))[:max_sessions]
+        files = _collect_session_files(sessions_dir, max_sessions)
         examples: list[dict] = []
 
         for fpath in files:
-            events = []
-            with open(fpath) as f:
-                for line in f:
-                    line = line.strip()
-                    if not line:
-                        continue
-                    try:
-                        events.append(json.loads(line))
-                    except json.JSONDecodeError:
-                        continue
+            events = _load_session_events(fpath)
 
             for example in _extract_retry_data(events):
-                example["metadata"]["session_id"] = (
-                    os.path.basename(fpath).replace(".jsonl", "")
-                )
+                example["metadata"]["session_id"] = Path(fpath).stem
                 example["metadata"]["source"] = "retry_session"
                 examples.append(example)
 
@@ -193,13 +183,9 @@ def export_retries(
 if __name__ == "__main__":
     import sys
 
-    from devloop.llmops.training import _default_sessions_dir
-
-    sessions_dir = _default_sessions_dir()
-    if not os.path.isdir(sessions_dir):
-        print(f"WARNING: Sessions dir not found: {sessions_dir}", file=sys.stderr)
-    elif not glob.glob(os.path.join(sessions_dir, "*.jsonl")):
-        print(f"WARNING: No .jsonl files in {sessions_dir}", file=sys.stderr)
+    files = _collect_session_files()
+    if not files:
+        print("WARNING: No session files found", file=sys.stderr)
 
     force = "--force" in sys.argv
     count = export_retries(force=force)