feat: add analyze_directory MCP tool — scan codebase, infer conventions, persist to .dervish/

2026-07-11 21:28:35 +02:00 · 2026-07-11 21:28:35 +02:00 · 830104b399
commit 830104b399
parent e94c52b71a
5 changed files with 349 additions and 6 deletions
--- a/AGENTS.md
+++ b/AGENTS.md
@ -41,11 +41,12 @@ python -m pytest tests/

 ## MCP Server

-The primary interface is an MCP server exposing a single tool:
+The primary interface is an MCP server exposing two tools:

 | Tool | Parameters | What it does |
 |------|-----------|-------------|
-| `infer_best_grammar` | `sequences`, `prefer`, `kmax`, `N`, `min_coverage` | Runs CRX + iDRegEx, picks best by MDL. `prefer='crx'` or `prefer='idregex'` skips ensemble. `min_coverage < 1.0` runs core+outlier analysis. |
+| `infer_best_grammar` | `sequences`, `prefer`, `kmax`, `N`, `min_coverage` | Infer grammar from raw sequences. Runs CRX + iDRegEx, picks best by MDL. |
+| `analyze_directory` | `directory`, `slice`, `min_coverage`, `prefer`, `kmax`, `include`, `exclude`, `main_only`, `max_mdl`, `persist` | Scan source code, infer conventions per package. Returns YAML grouped by module. Auto-persists to `{directory}/.dervish/grammars.yml`. |

 Start it: `python /path/to/bex/mcp_server.py`, then connect any MCP client.

@ -58,4 +59,4 @@ python -m bex.tag_preprocessor.analyze /path/to/codebase --verbose
 python -m bex.tag_preprocessor.analyze /path/to/codebase --slice package --include '**/src/**'
 ```

-Key flags: `--slice package` (per-directory grammars), `--verbose` (progress), `--include`/`--exclude` (glob filters), `--kore` (enable slow kORE in ensemble).
+Key flags: `--slice package` (per-directory grammars), `--verbose` (progress), `--include`/`--exclude` (glob filters), `--main-only` (exclude test files), `--kore` (enable slow kORE in ensemble).
--- a/README.md
+++ b/README.md
@ -50,9 +50,18 @@ The primary interface is a **Model Context Protocol (MCP)** server. Connect any

 | Tool | Parameters | What it does |
 |------|-----------|-------------|
-| `infer_best_grammar` | `sequences`, `prefer`, `kmax`, `N`, `min_coverage` | **The only tool you need.** Runs CRX + iDRegEx, picks best by MDL. Set `prefer` to run only one algorithm. Set `min_coverage < 1.0` for optional core+outlier analysis. |
+| `infer_best_grammar` | `sequences`, `prefer`, `kmax`, `N`, `min_coverage` | Infer grammar from raw sequences. Runs CRX + iDRegEx, picks best by MDL. Set `prefer` to run only one algorithm. Set `min_coverage < 1.0` for optional core+outlier analysis. |
+| `analyze_directory` | `directory`, `slice`, `min_coverage`, `prefer`, `kmax`, `include`, `exclude`, `main_only`, `max_mdl`, `persist` | Scan a source code directory and infer behavioral conventions per package. Returns YAML grouped by module. Auto-persists to `{directory}/.dervish/grammars.yml`. |

-**Parameters explained:**
+**`analyze_directory` parameters:**
+- **`directory`**: Path to source code directory to analyze.
+- **`slice`**: `'package'` (per-directory, default) or `'flat'` (one per language).
+- **`main_only`**: Exclude test files when `True`. Default `False`.
+- **`max_mdl`**: Drop groups with MDL above this threshold. Default 200. Lower = tighter patterns only.
+- **`persist`**: Write results to `{directory}/.dervish/grammars.yml`. Default `True`.
+- Other parameters same as `infer_best_grammar`.
+
+**Parameters explained for `infer_best_grammar`:**
 - **`prefer`**: `'crx'` for full vocabulary (accepts all sequences), `'idregex'` for deterministic minimal core, `'koreinference'` for k-OA with rwr₀ repair (slow, rarely wins). Omit to let MDL pick the winner across CRX and iDRegEx.
 - **`kmax`** (1–5): Context window for k-ORE inference (iDRegEx, kOREInference). Higher values capture longer-range dependencies but need more data and are slower. Default 2 works for most cases.
 - **`N`** (1–10): Random trials for k-ORE inference. More = better convergence but slower. Default 3.
--- a/bex/mcp_server.py
+++ b/bex/mcp_server.py
@ -7,6 +7,11 @@ Run as: python -m bex.mcp_server
 from mcp.server.fastmcp import FastMCP

 from .ensemble import infer_ensemble, _matches
+from .tag_preprocessor.analyze import (
+    analyze_directory as _analyze_directory,
+    _build_yaml_output,
+    _persist_grammars,
+)

 mcp = FastMCP("grammar-inference", log_level="ERROR")

@ -77,6 +82,69 @@ def infer_best_grammar(
    return "\n".join(lines)


+@mcp.tool()
+def analyze_directory(
+    directory: str,
+    slice: str = "package",
+    min_coverage: float = 0.8,
+    prefer: str = "",
+    kmax: int = 2,
+    include: str = "",
+    exclude: str = "",
+    main_only: bool = False,
+    max_mdl: float = 200,
+    persist: bool = True,
+) -> str:
+    """Scan a source code directory and infer behavioral conventions
+    (regular expression grammars) per package. Returns compact patterns
+    grouped by module, sorted by quality (MDL score).
+
+    Use this when you need to understand the calling conventions in a
+    codebase — what patterns new code should follow. The grammar
+    compresses each package's method call patterns into a compact
+    regular expression.
+
+    Auto-persists results to {directory}/.dervish/grammars.yml unless
+    persist=False.
+
+    Args:
+        directory: Path to the source code directory to analyze.
+        slice: Grouping strategy — 'package' (per directory, default)
+            or 'flat' (one per language).
+        min_coverage: BEX core coverage threshold for outlier removal
+            (0.5–1.0). Default 0.8.
+        prefer: Optional — 'crx' for full vocabulary, 'idregex' for
+            minimal core. Omit to auto-pick by MDL.
+        kmax: Context depth for k-ORE inference. Default 2.
+        include: Glob pattern to include only matching files.
+        exclude: Glob pattern to skip matching files.
+        main_only: When True, exclude test files (src/test/**, *Test.*,
+            etc.). Default False.
+        max_mdl: Drop groups with MDL above this threshold. Default 200.
+            Lower = tighter patterns only. Set higher to see noisier groups.
+        persist: When True (default), write results to
+            {directory}/.dervish/grammars.yml.
+
+    Returns:
+        YAML string with grammars grouped by top-level module, sorted
+        by MDL (tightest/most useful first).
+    """
+    results = _analyze_directory(
+        directory,
+        min_coverage=min_coverage,
+        prefer=prefer or None,
+        kmax=kmax,
+        slice=slice,
+        include=include or None,
+        exclude=exclude or None,
+        main_only=main_only,
+    )
+    yaml_content = _build_yaml_output(results, directory, max_mdl=max_mdl)
+    if persist:
+        _persist_grammars(yaml_content, directory)
+    return yaml_content
+
+
 def main():
    mcp.run()

--- a/bex/tag_preprocessor/analyze.py
+++ b/bex/tag_preprocessor/analyze.py
@ -31,6 +31,21 @@ FALLBACK_SKIP = {
    ".venv", "venv", "env", ".env", "out",
 }

+TEST_PATH_PATTERNS = [
+    "**/test/**", "**/tests/**", "**/*Test*/**",
+    "**/*_test.*", "**/*_spec.*", "**/test_*.py",
+    "**/*Test.kt", "**/*Test.java", "**/*Test.js",
+    "**/*Spec.*", "**/*_test.go",
+]
+
+
+def _is_main_source(filepath):
+    """Return True if file path looks like main source (not test)."""
+    for pattern in TEST_PATH_PATTERNS:
+        if _match_glob(filepath, pattern):
+            return False
+    return True
+

 def _load_gitignore(dir_path):
    """Load .gitignore from dir_path root, return PathSpec or None."""
@ -409,6 +424,7 @@ def analyze_directory(
    slice="flat",
    include=None,
    exclude=None,
+    main_only=False,
    include_kore=False,
 ):
    """Scan a directory and run analysis for each language found.
@ -421,6 +437,7 @@ def analyze_directory(
        slice: grouping strategy — "flat" (one per language) or "package" (per directory).
        include: optional glob — only process files matching this pattern.
        exclude: optional glob — skip files matching this pattern.
+        main_only: exclude test files when True.

    Returns:
        dict mapping extension → list of (label, result_dict, count, meta) tuples.
@ -431,6 +448,8 @@ def analyze_directory(
        if len(files) < 1:
            continue
        files = _filter_glob(files, include=include, exclude=exclude)
+        if main_only:
+            files = [f for f in files if _is_main_source(f)]
        if not files:
            continue
        if slice == "package":
@ -469,7 +488,7 @@ def _build_json_output(results):
            if result and result.get("best"):
                entry["algorithm"] = result["best"]["algorithm"]
                entry["grammar"] = result["best"]["grammar"]
-                entry["mdl_score"] = round(result["best"]["mdl_score"], 1)
+                entry["mdl_score"] = round(result['best']['mdl_score'], 1)
            entry["imports"] = meta.get("imports", [])
            entry["arg_patterns"] = meta.get("arg_patterns", {})
            lang["conventions"].append(entry)
@ -478,6 +497,70 @@ def _build_json_output(results):
    return json.dumps(output, indent=2)


+def _build_yaml_output(results, dir_path, max_mdl=200.0):
+    """Build YAML output grouped by top-level module, sorted by MDL.
+
+    Filters out (other), no-grammar groups, and groups above max_mdl.
+    Returns YAML string.
+    """
+    import yaml
+
+    project_name = os.path.basename(os.path.abspath(dir_path))
+
+    # Collect all entries grouped by top-level module
+    modules = {}
+    total_patterns = 0
+    total_methods = 0
+
+    for ext, clusters in results.items():
+        for label, result, count, meta in clusters:
+            total_methods += count
+            if label == "(other)":
+                continue
+            if not result or not result.get("best"):
+                continue
+            best = result["best"]
+            if best["mdl_score"] > max_mdl:
+                continue
+
+            # Extract top-level module from package path
+            parts = label.replace(os.sep, "/").split("/")
+            module = parts[0] if len(parts) > 1 else "(root)"
+
+            entry = {
+                "package": label,
+                "methods": count,
+                "grammar": best["grammar"],
+                "algorithm": best["algorithm"],
+                "mdl": round(best["mdl_score"], 1),
+            }
+
+            modules.setdefault(module, []).append(entry)
+            total_patterns += 1
+
+    # Sort entries within each module by MDL
+    for module in modules:
+        modules[module].sort(key=lambda x: x["mdl"])
+
+    # Sort modules by name
+    ordered = dict(sorted(modules.items()))
+
+    # Build YAML
+    header = f"# {project_name} — {total_patterns} patterns ({total_methods} methods)\n\n"
+    yaml_content = yaml.dump(ordered, default_flow_style=False, sort_keys=False, allow_unicode=True)
+    return header + yaml_content
+
+
+def _persist_grammars(yaml_content, dir_path):
+    """Write grammars to {dir_path}/.dervish/grammars.yml"""
+    dervish_dir = os.path.join(dir_path, ".dervish")
+    os.makedirs(dervish_dir, exist_ok=True)
+    out_path = os.path.join(dervish_dir, "grammars.yml")
+    with open(out_path, "w") as f:
+        f.write(yaml_content)
+    _vprint(f"Persisted to {out_path}")
+
+
 def _parse_args(argv=None):
    parser = argparse.ArgumentParser(
        description="Analyze a directory of source code for behavioral conventions.",
--- a/docs/plans/analyze-directory-mcp-tool.md
+++ b/docs/plans/analyze-directory-mcp-tool.md
@ -0,0 +1,182 @@
+# Plan: `analyze_directory` MCP Tool
+
+## Goal
+
+Add an `analyze_directory` MCP tool that scans a source code directory, infers behavioral conventions per package, returns YAML grouped by module, and auto-persists to `{directory}/.dervish/grammars.yml`.
+
+## Design Decisions (locked)
+
+| Decision | Choice |
+|----------|--------|
+| Signature | New separate tool (not extending `infer_best_grammar`) |
+| Persistence | Auto-save to `{directory}/.dervish/grammars.yml` (on by default, `persist=False` to opt out) |
+| Output format | YAML grouped by top-level module, sorted by MDL |
+| Heuristics | MDL threshold (default 200), drop `(other)`, drop no-grammar, optional `main_only` |
+| YAML library | pyyaml (`yaml.dump()`) |
+
+## Signature
+
+```python
+@mcp.tool()
+def analyze_directory(
+    directory: str,              # Path to source code directory
+    slice: str = "package",      # "flat" or "package"
+    min_coverage: float = 0.8,   # Outlier removal threshold
+    prefer: str = "",            # Force CRX or iDRegEx
+    kmax: int = 2,               # k-ORE context depth
+    include: str = "",           # Glob filter (include)
+    exclude: str = "",           # Glob filter (exclude)
+    main_only: bool = False,     # Exclude test code paths
+    max_mdl: float = 200,        # Drop groups above this MDL
+    persist: bool = True,        # Auto-save to .dervish/
+) -> str:
+```
+
+## Output format (YAML)
+
+```yaml
+# RAGSAK — 35 patterns (1594 methods, 462 files)
+
+agents:
+  - package: agents/rag/embabel
+    methods: 52
+    grammar: "info+"
+    algorithm: CRX
+    mdl: 28.0
+
+modules:
+  - package: modules/common/.../ids
+    methods: 18
+    grammar: "(of|requireSafeId)"
+    algorithm: iDRegEx
+    mdl: 34.0
+  - package: modules/ingestion/.../batch
+    methods: 16
+    grammar: "info?.StepBuilder?.listener+?.build"
+    algorithm: CRX
+    mdl: 13.2
+
+infrastructure:
+  - package: infrastructure/.../service/cleanup
+    methods: 4
+    grammar: "info.(deleteByJobId+deleteByKnowledgeBaseId)"
+    algorithm: CRX
+    mdl: 7.0
+```
+
+## Persistence
+
+- Write to `{directory}/.dervish/grammars.yml`
+- Create `.dervish/` dir if missing
+- Overwrite on each run (idempotent)
+- File is committed to git alongside the code
+- LLM reads it later without recomputing
+
+## Heuristics
+
+1. **MDL threshold** — drop groups with MDL > `max_mdl` (default 200)
+2. **Drop `(other)`** — ungrouped methods, always excluded
+3. **Drop no-grammar** — groups where both algorithms failed, always excluded
+4. **`main_only`** — excludes test paths when enabled
+
+Test path patterns for `main_only`:
+```
+**/test/**, **/tests/**, **/*Test*/**
+**/*_test.*, **/*_spec.*, **/test_*.py
+**/*Test.kt, **/*Test.java, **/*Test.js
+**/*Spec.*, **/*_test.go
+```
+
+## Files to modify
+
+| File | Change |
+|------|--------|
+| `bex/tag_preprocessor/analyze.py` | Add `_is_main_source()`, `_build_yaml_output()`, `_persist_grammars()`, wire `main_only` into `analyze_directory()` |
+| `bex/mcp_server.py` | Add `analyze_directory` MCP tool |
+| `README.md` | Document new tool in MCP tools table |
+| `AGENTS.md` | Document new tool in MCP tools table |
+
+## Implementation steps
+
+### Step 1: `_is_main_source()` in `analyze.py`
+
+```python
+TEST_PATH_PATTERNS = [
+    "**/test/**", "**/tests/**", "**/*Test*/**",
+    "**/*_test.*", "**/*_spec.*", "**/test_*.py",
+    "**/*Test.kt", "**/*Test.java", "**/*Test.js",
+    "**/*Spec.*", "**/*_test.go",
+]
+
+def _is_main_source(filepath):
+    """Return True if file path looks like main source (not test)."""
+    for pattern in TEST_PATH_PATTERNS:
+        if _match_glob(filepath, pattern):
+            return False
+    return True
+```
+
+### Step 2: `_build_yaml_output()` in `analyze.py`
+
+Convert results dict to YAML grouped by top-level module:
+
+```python
+def _build_yaml_output(results, dir_path):
+    """Build YAML output grouped by top-level module.
+    
+    Returns YAML string with grammar per package, sorted by MDL within each module.
+    Skips (other) and no-grammar groups.
+    """
+```
+
+Logic:
+- Extract top-level module from package path (first directory component)
+- Sort groups by MDL within each module
+- Skip `(other)` and no-grammar groups
+- Return `yaml.dump()` output
+
+### Step 3: `_persist_grammars()` in `analyze.py`
+
+```python
+def _persist_grammars(yaml_content, dir_path):
+    """Write grammars to {dir_path}/.dervish/grammars.yml"""
+    dervish_dir = os.path.join(dir_path, ".dervish")
+    os.makedirs(dervish_dir, exist_ok=True)
+    with open(os.path.join(dervish_dir, "grammars.yml"), "w") as f:
+        f.write(yaml_content)
+```
+
+### Step 4: Wire `analyze_directory()` in `analyze.py`
+
+- Add `main_only` parameter
+- When `main_only=True`, filter files through `_is_main_source()` before processing
+- Apply MDL threshold filtering after inference
+- Return filtered results
+
+### Step 5: Add MCP tool in `mcp_server.py`
+
+```python
+@mcp.tool()
+def analyze_directory(...) -> str:
+```
+
+Flow:
+1. Call `analyze_directory()` from `tag_preprocessor.analyze`
+2. Filter by `max_mdl`, drop `(other)`, drop no-grammar
+3. Build YAML via `_build_yaml_output()`
+4. If `persist=True`, call `_persist_grammars()`
+5. Return YAML string
+
+### Step 6: Update docs
+
+- README.md: Add `analyze_directory` to MCP tools table
+- AGENTS.md: Add `analyze_directory` to MCP tools table
+
+## Verification
+
+1. Run: `python -m bex.mcp_server` (starts MCP server)
+2. Call `analyze_directory(directory="/home/tobi/Desktop/kesai/RAGSAK")`
+3. Verify YAML output is grouped by module, sorted by MDL
+4. Verify `.dervish/grammars.yml` created in RAGSAK
+5. Run existing tests: `python -m pytest tests/`
+6. Run: `python -m bex.tag_preprocessor.analyze /home/tobi/Desktop/kesai/RAGSAK --slice package --verbose` (verify no regression)