feat: Codex session transcript ingestion into BrainLayer (#83)

- New adapter: src/brainlayer/ingest/codex.py — parse, classify, chunk, embed, store
- CLI command: brainlayer ingest-codex <file|dir>
- Smart mapping: function_call_output → file_read/stack_trace/git_diff
- Batch ingest with dedup (skips already-indexed sessions)
- Fixed source field propagation in index_new.py
- 39 tests, validated on real data: 4,129 chunks from 20 sessions

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

Author: EtanHey

scripts/cloud_backfill.py

@@ -624,17 +624,26 @@ def poll_gemini_batch(batch_name: str, timeout_hours: float = 25) -> Dict[str, A
 
     print(f"\nPolling job: {batch_name}")
     while time.time() < deadline:
-        batch_job = client.batches.get(name=batch_name)
+        try:
+            batch_job = client.batches.get(name=batch_name)
+        except Exception as exc:
+            err_str = str(exc)
+            if "404" in err_str or "NOT_FOUND" in err_str:
+                print(f"  Job not found (404) — may have expired: {batch_name}")
+                return {"state": "failed", "error": "NOT_FOUND (expired or invalid)"}
+            raise
+        # state may be "JOB_STATE_SUCCEEDED" or "JobState.JOB_STATE_SUCCEEDED"
         state = str(batch_job.state)
+        state_normalized = state.split(".")[-1]  # strip "JobState." prefix if present
 
-        if state == "JOB_STATE_SUCCEEDED":
+        if state_normalized == "JOB_STATE_SUCCEEDED":
             print("  Job SUCCEEDED!")
             return {"state": "succeeded", "job": batch_job}
-        elif state == "JOB_STATE_FAILED":
+        elif state_normalized == "JOB_STATE_FAILED":
             error = getattr(batch_job, "error", "unknown error")
             print(f"  Job FAILED: {error}")
             return {"state": "failed", "error": str(error), "job": batch_job}
-        elif state == "JOB_STATE_CANCELLED":
+        elif state_normalized == "JOB_STATE_CANCELLED":
             print("  Job CANCELLED")
             return {"state": "failed", "error": "cancelled"}
 
@@ -674,8 +683,8 @@ def download_gemini_results(batch_job) -> List[Dict[str, Any]]:
     print(f"  Downloading results: {file_name}")
 
     try:
-        # Download the result file
-        content = client.files.download(name=file_name)
+        # Download the result file (SDK uses file= not name=)
+        content = client.files.download(file=file_name)
         if isinstance(content, bytes):
             content = content.decode("utf-8")
 
@@ -963,13 +972,24 @@ def resume_backfill(db_path: Path) -> None:
             return
 
         print(f"Found {len(pending)} pending jobs to resume")
-        for job in pending:
-            print(f"\nResuming: {job['batch_id']}")
-            result = poll_gemini_batch(job["batch_id"])
+        imported_count = 0
+        failed_count = 0
+        for i, job in enumerate(pending):
+            print(f"\nResuming [{i+1}/{len(pending)}]: {job['batch_id']}")
+            try:
+                result = poll_gemini_batch(job["batch_id"])
+            except Exception as exc:
+                print(f"  ERROR polling job: {exc}")
+                save_checkpoint(store, batch_id=job["batch_id"], status="failed",
+                                error=str(exc)[:500],
+                                completed_at=datetime.now(timezone.utc).isoformat())
+                failed_count += 1
+                continue
 
             if result["state"] == "succeeded":
                 batch_results = download_gemini_results(result["job"])
                 import_results(store, batch_results, job["batch_id"])
+                imported_count += 1
 
                 # Log usage to Supabase (best-effort)
                 batch_job = result.get("job")
@@ -986,9 +1006,11 @@ def resume_backfill(db_path: Path) -> None:
                 save_checkpoint(store, batch_id=job["batch_id"], status="failed",
                                 error=result.get("error", "unknown"),
                                 completed_at=datetime.now(timezone.utc).isoformat())
+                failed_count += 1
 
         final_stats = store.get_enrichment_stats()
-        print(f"\nEnrichment: {final_stats['enriched']}/{final_stats['total_chunks']} ({final_stats['percent']}%)")
+        print(f"\nImported {imported_count} jobs, {failed_count} failed")
+        print(f"Enrichment: {final_stats['enriched']}/{final_stats['total_chunks']} ({final_stats['percent']}%)")
 
     finally:
         store.close()

src/brainlayer/cli/__init__.py

@@ -5,6 +5,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Optional
 
 import typer
 from rich import print as rprint
@@ -1543,6 +1544,69 @@ def progress_callback(embedded_count, total_embed):
         raise typer.Exit(1)
 
 
+@app.command("ingest-codex")
+def ingest_codex(
+    path: Optional[Path] = typer.Argument(
+        None,
+        help="Codex session JSONL file or sessions directory (default: ~/.codex/sessions)",
+    ),
+    project: str = typer.Option(None, "--project", "-p", help="Override project name"),
+    since_days: int = typer.Option(None, "--since-days", "-d", help="Only process last N days"),
+    dry_run: bool = typer.Option(False, "--dry-run", help="Parse but do not write to DB"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="Print each classified entry"),
+) -> None:
+    """Ingest Codex (GPT-5.4) session transcripts into BrainLayer.
+
+    Normalizes ~/.codex/sessions/YYYY/MM/DD/*.jsonl into searchable chunks
+    with source='codex_cli'. Already-indexed sessions are skipped automatically.
+    """
+    from ..ingest.codex import ingest_codex_dir, ingest_codex_session
+    from ..paths import DEFAULT_DB_PATH
+
+    db_path = DEFAULT_DB_PATH
+
+    try:
+        if path and path.is_file():
+            rprint(f"[bold blue]זיכרון[/] — Ingesting Codex session: [bold]{path.name}[/]")
+            with console.status("Indexing..."):
+                n = ingest_codex_session(
+                    path,
+                    db_path=db_path,
+                    project_override=project,
+                    dry_run=dry_run,
+                    verbose=verbose,
+                )
+            rprint(f"[bold green]✓[/] Indexed [bold]{n}[/] chunks from {path.name}")
+        else:
+            sessions_root = path if path else None
+            label = str(sessions_root) if sessions_root else "~/.codex/sessions"
+            rprint(f"[bold blue]זיכרון[/] — Ingesting Codex sessions from: [bold]{label}[/]")
+            with Progress(
+                SpinnerColumn(),
+                TextColumn("[progress.description]{task.description}"),
+                TimeElapsedColumn(),
+                console=console,
+            ) as progress:
+                task = progress.add_task("Scanning sessions...", total=None)
+                files, chunks = ingest_codex_dir(
+                    sessions_dir=sessions_root,
+                    db_path=db_path,
+                    project_override=project,
+                    since_days=since_days,
+                    dry_run=dry_run,
+                    verbose=verbose,
+                )
+                progress.update(task, description=f"Done — {files} files, {chunks:,} chunks")
+            tag = " [dim](dry run)[/]" if dry_run else ""
+            rprint(f"[bold green]✓[/] Processed [bold]{files}[/] session files, [bold]{chunks:,}[/] chunks{tag}")
+    except FileNotFoundError as e:
+        rprint(f"[bold red]Error:[/] {e}")
+        raise typer.Exit(1)
+    except Exception as e:
+        rprint(f"[bold red]Error:[/] {e}")
+        raise typer.Exit(1)
+
+
 @app.command("analyze-semantic")
 def analyze_semantic(
     whatsapp_limit: int = typer.Option(5000, "--whatsapp-limit", "-w", help="Number of WhatsApp messages to analyze"),

src/brainlayer/index_new.py

@@ -79,6 +79,7 @@ def index_chunks_to_sqlite(
                 "conversation_id": conversation_id,
                 "position": i,
                 "sender": chunk.metadata.get("sender"),
+                "source": chunk.metadata.get("source", "claude_code"),
             }
         )
 

src/brainlayer/ingest/__init__.py

@@ -0,0 +1 @@
+"""Ingest adapters for non-Claude AI session transcripts."""