Merge pull request #122 from Ar9av/feat/batch-extract

feat(batch): parallel subagent batch planning for large-folder ingestion

Author: Ar9av

.skills/wiki-ingest/SKILL.md

@@ -94,6 +94,33 @@ This keeps faith with the "immutable raw layer" principle in `llm-wiki/SKILL.md`
 
 ## The Ingest Process
 
+### Step 0: Batch Planning for Large Folders
+
+**GUARD: Only run this step when the source is a directory with more than 20 files.** For single files, small folders, or `_raw/` mode, skip directly to Step 1.
+
+When the source is a large directory of docs, plan the parallel dispatch first:
+
+```bash
+obsidian-wiki batch-plan "$OBSIDIAN_VAULT_PATH" <source-dir> --pretty
+```
+
+This outputs a JSON plan with `batches` (each a list of files + total_bytes + kind counts) and `stats` (total, to_ingest, skipped_unchanged).
+
+**What to do with the plan:**
+
+1. **Check `stats.skipped_unchanged`** — report to the user how many files are being skipped (already ingested, hash unchanged).
+2. **If `batch_count == 0`** — all files are unchanged. Tell the user and stop.
+3. **If `batch_count == 1`** — proceed with the single batch as a normal Step 1 ingest.
+4. **If `batch_count > 1`** — dispatch each batch as a **parallel subagent** (multiple Agent tool calls in a single message). Each subagent receives a message like:
+   ```
+   Ingest these files into the wiki at $OBSIDIAN_VAULT_PATH using wiki-ingest Step 1 onward:
+   <list of file paths from this batch>
+   Skip batch-plan — these files are already partitioned.
+   ```
+   Wait for all subagents to complete, then run `/cross-linker` once to wire cross-references across all batches.
+
+**Fallback** (if `obsidian-wiki` is not installed): process files sequentially in groups of 15.
+
 ### Step 1: Read the Source
 
 Read the source(s) the user wants to ingest. In append mode, skip files the manifest says are already ingested and unchanged. Supported formats:

obsidian_wiki/batch.py

@@ -0,0 +1,241 @@
+"""Batch planner for parallel wiki-ingest subagent dispatch.
+
+When ingesting a large folder of docs, this module splits the source list into
+batches and emits a dispatch plan the skill uses to spawn parallel Claude
+subagents — each handling one batch independently, then merging results.
+
+The agent calls `obsidian-wiki batch-plan <vault> <source-dir> [options]`
+and gets back a JSON plan:
+
+{
+  "batches": [
+    {
+      "id": 0,
+      "files": ["path/to/a.md", "path/to/b.pdf"],
+      "total_bytes": 45000,
+      "kinds": {"markdown": 1, "pdf": 1}
+    },
+    ...
+  ],
+  "stats": {
+    "total_files": N,
+    "total_bytes": N,
+    "batch_count": N,
+    "skipped_unchanged": N,
+    "skipped_binary": N
+  },
+  "merge_hint": "Run /wiki-ingest on each batch in parallel, then run /cross-linker once all batches are done."
+}
+
+The skill dispatches each batch as a parallel subagent call, then runs
+cross-linker once all agents report completion.
+"""
+
+from __future__ import annotations
+
+import mimetypes
+import os
+from pathlib import Path
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# File classification
+# ---------------------------------------------------------------------------
+
+# Ingestible text/doc extensions
+TEXT_EXTENSIONS = frozenset(
+    ".md .mdx .txt .rst .html .htm .csv .tsv .json .jsonl .yaml .yml .xml".split()
+)
+PDF_EXTENSIONS = frozenset(".pdf".split())
+IMAGE_EXTENSIONS = frozenset(".png .jpg .jpeg .webp .gif .bmp .tiff .svg".split())
+OFFICE_EXTENSIONS = frozenset(".docx .xlsx .pptx .odt .ods .odp".split())
+CODE_EXTENSIONS = frozenset(
+    ".py .ts .js .jsx .tsx .go .rs .java .kt .rb .c .cpp .h .hpp .swift .sh".split()
+)
+
+# Binary / generated — skip entirely
+SKIP_EXTENSIONS = frozenset(
+    ".pyc .pyo .pyd .so .dylib .dll .exe .class .jar .war .egg "
+    ".zip .tar .gz .bz2 .whl .lock .mp4 .mov .mp3 .wav .ttf .woff .eot".split()
+)
+
+SKIP_DIRS = frozenset(
+    "node_modules .git __pycache__ .pytest_cache dist build target "
+    ".venv venv env .mypy_cache .ruff_cache coverage .tox .obsidian "
+    "_raw _archived _staging _archives".split()
+)
+
+
+def _classify(path: Path) -> str:
+    ext = path.suffix.lower()
+    if ext in TEXT_EXTENSIONS:
+        return "text"
+    if ext in PDF_EXTENSIONS:
+        return "pdf"
+    if ext in IMAGE_EXTENSIONS:
+        return "image"
+    if ext in OFFICE_EXTENSIONS:
+        return "office"
+    if ext in CODE_EXTENSIONS:
+        return "code"
+    if ext in SKIP_EXTENSIONS:
+        return "skip"
+    # Guess via mime
+    mime, _ = mimetypes.guess_type(str(path))
+    if mime and mime.startswith("text/"):
+        return "text"
+    return "skip"
+
+
+def _file_size(path: Path) -> int:
+    try:
+        return path.stat().st_size
+    except OSError:
+        return 0
+
+
+# ---------------------------------------------------------------------------
+# Source discovery
+# ---------------------------------------------------------------------------
+
+def discover_sources(
+    source_dir: Path,
+    *,
+    vault: Path | None = None,
+    include_code: bool = False,
+) -> list[dict]:
+    """Walk source_dir and return a list of ingestible file dicts.
+
+    Each dict: {path, kind, size_bytes}. Code files are excluded by default
+    because wiki-ingest Step 1c handles them via ast-extract separately.
+    """
+    files = []
+    for dirpath, dirnames, filenames in os.walk(source_dir):
+        dirnames[:] = [
+            d for d in dirnames
+            if d not in SKIP_DIRS and not d.startswith(".")
+        ]
+        for fn in sorted(filenames):
+            p = Path(dirpath) / fn
+            kind = _classify(p)
+            if kind == "skip":
+                continue
+            if kind == "code" and not include_code:
+                continue
+            files.append({"path": str(p), "kind": kind, "size_bytes": _file_size(p)})
+    return files
+
+
+# ---------------------------------------------------------------------------
+# Manifest filtering (skip unchanged sources)
+# ---------------------------------------------------------------------------
+
+def _filter_unchanged(
+    files: list[dict],
+    vault: Path,
+) -> tuple[list[dict], int]:
+    """Remove files whose hash matches the manifest. Returns (to_ingest, skipped_count)."""
+    try:
+        from obsidian_wiki.cache import check_sources, compute_hash
+        paths = [Path(f["path"]) for f in files]
+        result = check_sources(vault, paths)
+        unchanged_set = set(result["unchanged"])
+        to_ingest = [f for f in files if f["path"] not in unchanged_set]
+        return to_ingest, len(unchanged_set)
+    except Exception:
+        return files, 0
+
+
+# ---------------------------------------------------------------------------
+# Batch splitting
+# ---------------------------------------------------------------------------
+
+def _make_batches(
+    files: list[dict],
+    *,
+    max_batch_bytes: int,
+    max_batch_files: int,
+) -> list[list[dict]]:
+    """Split files into batches respecting size and file-count limits."""
+    batches: list[list[dict]] = []
+    current: list[dict] = []
+    current_bytes = 0
+
+    for f in files:
+        sz = f["size_bytes"]
+        if current and (
+            current_bytes + sz > max_batch_bytes
+            or len(current) >= max_batch_files
+        ):
+            batches.append(current)
+            current = []
+            current_bytes = 0
+        current.append(f)
+        current_bytes += sz
+
+    if current:
+        batches.append(current)
+    return batches
+
+
+# ---------------------------------------------------------------------------
+# Main entry point
+# ---------------------------------------------------------------------------
+
+def plan_batches(
+    source_dir: Path,
+    vault: Path,
+    *,
+    max_batch_mb: float = 2.0,
+    max_batch_files: int = 20,
+    skip_unchanged: bool = True,
+    include_code: bool = False,
+) -> dict[str, Any]:
+    """Discover sources, filter unchanged, and split into batches."""
+    all_files = discover_sources(source_dir, vault=vault, include_code=include_code)
+
+    skipped_binary = 0  # files already excluded by _classify
+    skipped_unchanged = 0
+
+    to_ingest = all_files
+    if skip_unchanged and vault.is_dir():
+        to_ingest, skipped_unchanged = _filter_unchanged(all_files, vault)
+
+    max_batch_bytes = int(max_batch_mb * 1024 * 1024)
+    batches_raw = _make_batches(
+        to_ingest,
+        max_batch_bytes=max_batch_bytes,
+        max_batch_files=max_batch_files,
+    )
+
+    total_bytes = sum(f["size_bytes"] for f in to_ingest)
+
+    batches_out = []
+    for i, batch in enumerate(batches_raw):
+        kinds: dict[str, int] = {}
+        for f in batch:
+            kinds[f["kind"]] = kinds.get(f["kind"], 0) + 1
+        batches_out.append({
+            "id": i,
+            "files": [f["path"] for f in batch],
+            "total_bytes": sum(f["size_bytes"] for f in batch),
+            "kinds": kinds,
+        })
+
+    merge_hint = (
+        "Dispatch each batch as a parallel subagent with /wiki-ingest on its file list. "
+        "Once all batches complete, run /cross-linker to wire up cross-references."
+    )
+
+    return {
+        "batches": batches_out,
+        "stats": {
+            "total_files": len(all_files),
+            "to_ingest": len(to_ingest),
+            "total_bytes": total_bytes,
+            "batch_count": len(batches_out),
+            "skipped_unchanged": skipped_unchanged,
+            "skipped_binary": skipped_binary,
+        },
+        "merge_hint": merge_hint,
+    }

obsidian_wiki/cli.py

@@ -349,6 +349,28 @@ def cmd_setup(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_batch_plan(args: argparse.Namespace) -> int:
+    from obsidian_wiki.batch import plan_batches
+    source_dir = Path(args.source_dir).expanduser().resolve()
+    vault = Path(args.vault).expanduser().resolve()
+    if not source_dir.is_dir():
+        print(f"error: source directory not found: {source_dir}", file=sys.stderr)
+        return 1
+    result = plan_batches(
+        source_dir,
+        vault,
+        max_batch_mb=args.max_mb,
+        max_batch_files=args.max_files,
+        skip_unchanged=not args.no_cache,
+        include_code=args.include_code,
+    )
+    if args.pretty:
+        print(json.dumps(result, indent=2))
+    else:
+        print(json.dumps(result))
+    return 0
+
+
 def cmd_graph_analyse(args: argparse.Namespace) -> int:
     from obsidian_wiki.graph_analysis import analyse_vault
     vault = Path(args.vault).expanduser().resolve()
@@ -469,6 +491,19 @@ def build_parser() -> argparse.ArgumentParser:
     ip = sub.add_parser("info", help="show install paths, version, and config")
     ip.set_defaults(func=cmd_info)
 
+    bp = sub.add_parser(
+        "batch-plan",
+        help="split a source directory into parallel-ingest batches, skipping unchanged files",
+    )
+    bp.add_argument("vault", help="path to the Obsidian vault")
+    bp.add_argument("source_dir", help="directory of source documents to ingest")
+    bp.add_argument("--max-mb", type=float, default=2.0, help="max MB per batch (default: 2)")
+    bp.add_argument("--max-files", type=int, default=20, help="max files per batch (default: 20)")
+    bp.add_argument("--no-cache", action="store_true", help="disable manifest-based skip of unchanged files")
+    bp.add_argument("--include-code", action="store_true", help="include code files (default: excluded; use ast-extract instead)")
+    bp.add_argument("--pretty", action="store_true", help="pretty-print JSON output")
+    bp.set_defaults(func=cmd_batch_plan)
+
     ga = sub.add_parser(
         "graph-analyse",
         help="analyse the vault's wikilink graph: god nodes, communities, surprising connections",