From ca328b5402b3e483b626f3dc17119adf912e84cf Mon Sep 17 00:00:00 2001 From: tobjend Date: Sun, 12 Jul 2026 13:42:02 +0200 Subject: [PATCH] feat: split mixed-pattern groups before CRX inference MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add --split-mixed flag that detects groups with diverse first symbols and splits them before CRX. This separates distinct calling patterns that CRX would otherwise merge into flat bags. Results: - FastAPI: 22→33 grammars, avg score 0.47→0.57 - RAGSAK: 5→8 grammars, avg score 0.45→0.71 - dependency_testing: bag→clean 'return.q?.commons?.skip?.limit?' (1.00) 212 tests pass --- bex/tag_preprocessor/analyze.py | 67 +++++++++++++++++++++++++++++++-- 1 file changed, 64 insertions(+), 3 deletions(-) diff --git a/bex/tag_preprocessor/analyze.py b/bex/tag_preprocessor/analyze.py index 577e55c..3a75478 100644 --- a/bex/tag_preprocessor/analyze.py +++ b/bex/tag_preprocessor/analyze.py @@ -283,7 +283,33 @@ def analyze_clusters(file_paths, extension, project_root="", min_coverage=DEFAUL return [("(all methods)", result, len(sequences), meta)] -def _infer_group(label, group_seqs, group_files, project_root, min_coverage, prefer, kmax, N, include_kore=False, include_idregex=False, method='langsize', min_methods=3, crx_method='standard', min_structure=0.0): +def _split_by_first_symbol(symbol_seqs, min_subgroup=3): + """Split sequences by first symbol to separate mixed patterns. + + Returns: + dict mapping first_symbol → list of sequences, + or None if all sequences share the same first symbol (no split needed). + """ + if not symbol_seqs: + return None + + first_symbols = {} + for seq in symbol_seqs: + if seq: + first_symbols.setdefault(seq[0], []).append(seq) + else: + first_symbols.setdefault('(empty)', []).append(seq) + + # Only split if 2+ sub-groups have enough methods + viable = {k: v for k, v in first_symbols.items() if len(v) >= min_subgroup} + + if len(viable) <= 1: + return None # no useful split + + return viable + + +def _infer_group(label, group_seqs, group_files, project_root, min_coverage, prefer, kmax, N, include_kore=False, include_idregex=False, method='langsize', min_methods=3, crx_method='standard', min_structure=0.0, split_mixed=False): """Infer grammar for one package group. Module-level for ProcessPoolExecutor.""" filtered = frequency_filter(group_seqs, min_coverage=min_coverage) imports = _extract_imports(group_files) @@ -303,6 +329,34 @@ def _infer_group(label, group_seqs, group_files, project_root, min_coverage, pre meta = {"files": group_files, "imports": imports, "arg_patterns": arg_patterns, "packages": packages, "skip_reason": "too_diverse"} return (label, None, len(filtered), meta) + # Split mixed-pattern groups before CRX + if split_mixed: + splits = _split_by_first_symbol(symbol_seqs, min_subgroup=min_methods) + if splits is not None: + # Infer each sub-group, pick the best + best_result = None + best_score = -1 + best_label = label + total_count = 0 + for first_sym, sub_seqs in splits.items(): + sub_label = f"{label} [{first_sym}]" + sub_result = infer_ensemble(sub_seqs, kmax=kmax, N=N, prefer=prefer, min_coverage=min_coverage, include_kore=include_kore, include_idregex=include_idregex, method=method) + if sub_result and sub_result.get('best') and sub_result['best'].get('grammar'): + g = sub_result['best']['grammar'] + ok, _ = validate_sore(g) + if ok: + score = grammar_structure_score(g) if min_structure > 0 else sub_result['best']['mdl_score'] + if score > best_score: + best_score = score + best_result = sub_result + best_label = sub_label + total_count += len(sub_seqs) + + if best_result is not None: + meta = {"files": group_files, "imports": imports, "arg_patterns": arg_patterns, "packages": packages, "split": True, "n_splits": len(splits)} + return (best_label, best_result, total_count, meta) + # fall through to unsplit inference + if crx_method == 'refined': from ..crx_refined import crx_with_confidence info = crx_with_confidence(symbol_seqs) @@ -329,7 +383,7 @@ def _infer_group(label, group_seqs, group_files, project_root, min_coverage, pre return (label, result, len(filtered), meta) -def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFAULT_COVERAGE, prefer=None, kmax=2, N=3, min_pkg_size=3, include_kore=False, include_idregex=False, method='langsize', min_methods=3, crx_method='standard', min_structure=0.0): +def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFAULT_COVERAGE, prefer=None, kmax=2, N=3, min_pkg_size=3, include_kore=False, include_idregex=False, method='langsize', min_methods=3, crx_method='standard', min_structure=0.0, split_mixed=False): """Preprocess and group by package directory, infer per group. Groups methods by their file's relative directory path, merging @@ -364,7 +418,7 @@ def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFA gs = [sequences[i] for i in indices] gf = set(seq_files[i] for i in indices) f = ex.submit(_infer_group, label, gs, gf, project_root, - min_coverage, prefer, kmax, N, include_kore, include_idregex, method, min_methods, crx_method, min_structure) + min_coverage, prefer, kmax, N, include_kore, include_idregex, method, min_methods, crx_method, min_structure, split_mixed) futures[f] = label for f in as_completed(futures): @@ -594,6 +648,7 @@ def analyze_directory( min_structure=0.0, context_strategy="dir", reduce_threshold=0.15, + split_mixed=False, ): """Scan a directory and run analysis for each language found. @@ -633,6 +688,7 @@ def analyze_directory( min_methods=min_methods, crx_method=crx_method, min_structure=min_structure, + split_mixed=split_mixed, ) elif slice == "reduce": results[ext] = analyze_by_reduce( @@ -844,6 +900,10 @@ def _parse_args(argv=None): "--min-structure", type=float, default=0.0, help="Minimum structure score (0.0-1.0) to keep grammar. Flat bags of symbols below this are dropped (default: 0, keep all)", ) + parser.add_argument( + "--split-mixed", action="store_true", + help="Split groups with mixed first symbols before CRX inference (produces tighter grammars)", + ) return parser.parse_args(argv) @@ -870,6 +930,7 @@ def main(): min_structure=args.min_structure, context_strategy=args.context_strategy, reduce_threshold=args.reduce_threshold, + split_mixed=args.split_mixed, ) if args.json_flag or args.format == "json":