feat: per-call argument pattern extraction

- _classify_arg_node: classify AST args as var/lit/call/lambda/kwarg/expr/other
- _find_arglist_node: language-agnostic arglist detection (named field →
  argument_list/call_suffix/template_string fallback)
- _iterate_arg_nodes: extract arg expressions from call_suffix (Kotlin lambdas),
  value_arguments (Kotlin parens), and standard argument lists
- extract_arg_info: per-file arg structure for all behavioral captures
- _summarize_arg_info: merge across files into min/max/common arg counts
- _build_arg_patterns: cluster-scoped aggregation
- Output shows top-4 calls per cluster with arg count + type patterns
- e.g. assertEquals: n=2 [lit,var], every: n=1 [lambda], verify: n=0-1 [lambda|var]

bex/tag_preprocessor/analyze.py

@@ -16,7 +16,7 @@ from collections import Counter
 
 import pathspec
 
-from .code import preprocess_by_method, _extract_call_tokens
+from .code import preprocess_by_method, _extract_call_tokens, extract_arg_info, _summarize_arg_info
 from bex.ensemble import infer_ensemble
 
 IMPORT_PATTERNS = [
@@ -58,6 +58,21 @@ def _match_glob(filepath, pattern):
     return spec.match_file(filepath)
 
 
+def _build_arg_patterns(file_paths):
+    """Extract merged argument patterns across a set of files."""
+    merged = {}
+    for fp in file_paths:
+        try:
+            with open(fp) as f:
+                code = f.read()
+        except OSError:
+            continue
+        info = extract_arg_info(fp, code)
+        for call_name, observations in info.items():
+            merged.setdefault(call_name, []).extend(observations)
+    return _summarize_arg_info(merged)
+
+
 def _extract_imports(file_paths):
     """Extract unique import lines from source files.
 
@@ -227,9 +242,10 @@ def analyze_clusters(file_paths, extension, min_coverage=0.2, prefer=None, kmax=
             idx = next(i for i, s in enumerate(sequences) if s is seq)
             cluster_fps.add(seq_files[idx])
         imports = _extract_imports(cluster_fps)
+        arg_patterns = _build_arg_patterns(cluster_fps)
         symbol_seqs = [[text for _, text, _ in seq] for seq in cluster_seqs]
         result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer)
-        meta = {"files": cluster_fps, "imports": imports}
+        meta = {"files": cluster_fps, "imports": imports, "arg_patterns": arg_patterns}
         results.append((label, result, len(cluster_seqs), meta))
 
     return results
@@ -365,6 +381,7 @@ def _build_json_output(results):
                 entry["grammar"] = result["best"]["grammar"]
                 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)
         lang["total_methods"] = total_methods
         output.append(lang)
@@ -402,6 +419,15 @@ def main():
                 print(f"     Imports:  {joined}")
                 if len(imps) > 6:
                     print(f"               ... and {len(imps) - 6} more")
+            argp = meta.get("arg_patterns", {})
+            if argp:
+                top_calls = sorted(argp.items(), key=lambda x: -x[1]["occurrences"])[:4]
+                for call_name, asum in top_calls:
+                    ac = asum["arg_count"]
+                    pats = asum["patterns"][:2]
+                    pat_strs = [f"{p['args']}:{','.join(p['types'])}" for p in pats]
+                    print(f"     Args({call_name}): n={ac['common']} "
+                          f"[{'; '.join(pat_strs)}]")
 
 
 if __name__ == "__main__":

bex/tag_preprocessor/code.py

@@ -72,6 +72,135 @@ def _extract_call_tokens(seq):
 _grammar_cache = {}
 _query_cache = {}
 
+ARG_LITERAL_TYPES = {
+    "string", "string_literal", "number", "integer", "float",
+    "decimal", "true", "false", "null", "nil", "none",
+}
+
+LAMBDA_TYPES = {
+    "lambda", "do_block", "block",
+}
+
+
+def _classify_arg_node(node):
+    """Classify a tree-sitter argument node by structural role."""
+    t = node.type
+    if t in ARG_LITERAL_TYPES:
+        return "lit"
+    if t in LAMBDA_TYPES or "ambda" in t or "block" in t:
+        return "lambda"
+    if t == "keyword_argument" or t.endswith("named_argument"):
+        return "kwarg"
+    if t.endswith("call_expression") or t in ("call", "method_invocation"):
+        return "call"
+    if t.endswith("identifier") or t.endswith("name"):
+        return "var"
+    if "subscript" in t:
+        return "subscript"
+    if "binary" in t or "unary" in t or "ternary" in t or "operator" in t:
+        return "expr"
+    if t in ("interpolation", "template_string"):
+        return "template"
+    return "other"
+
+
+def _find_arglist_node(parent):
+    """Find the argument container node in a call expression, language-agnostic."""
+    args = parent.child_by_field_name("arguments")
+    if args:
+        return args
+    for child in parent.children:
+        if child.type in ("argument_list", "arguments", "call_suffix"):
+            return child
+        if child.type == "template_string":
+            return child
+    return None
+
+
+def _iterate_arg_nodes(arglist):
+    """Yield individual argument expression nodes from an arglist."""
+    if arglist.type == "call_suffix":
+        for child in arglist.children:
+            if child.is_named and ("ambda" in child.type or "block" in child.type):
+                yield child
+                return
+        for child in arglist.children:
+            if child.type == "value_arguments":
+                for arg in child.children:
+                    if arg.is_named and arg.type == "value_argument":
+                        for expr in arg.children:
+                            if expr.is_named:
+                                yield expr
+                                break
+        return
+    for child in arglist.children:
+        if child.is_named:
+            yield child
+
+
+def extract_arg_info(file_path, code):
+    """For each behavioral call in the file, record argument structure.
+
+    Returns dict: call_text -> [ (arg_count, (type1, type2, ...)), ... ]
+    where types are 'var','lit','call','lambda','kwarg','expr','other'.
+    """
+    ext = os.path.splitext(file_path)[1].lower()
+    lang, query_name = _load_grammar(ext)
+    query_src = _load_query(query_name)
+
+    parser = Parser(lang)
+    tree = parser.parse(code.encode())
+    query = Query(lang, query_src)
+
+    cursor = QueryCursor(query)
+    captures = cursor.captures(tree.root_node)
+
+    info = {}
+    for capname, nodes in captures.items():
+        if not capname.startswith(BEHAVIORAL_PREFIXES):
+            continue
+        for node in nodes:
+            text = code[node.start_byte:node.end_byte].strip()
+            parent = node.parent
+            if not parent:
+                continue
+            arglist = _find_arglist_node(parent)
+            if not arglist:
+                continue
+            types = [_classify_arg_node(c) for c in _iterate_arg_nodes(arglist)]
+            info.setdefault(text, []).append((len(types), tuple(types)))
+    return info
+
+
+def _summarize_arg_info(info):
+    """Aggregate extract_arg_info across files into compact summaries.
+
+    Returns dict: call_text -> {
+        occurrences: int,
+        arg_count: {min, max, common},
+        patterns: [(count, (types..)), ...],  # top-5 by frequency
+    }
+    """
+    from collections import Counter
+    summary = {}
+    for call_name, observations in info.items():
+        counts = [c for c, _ in observations]
+        pattern_counts = Counter(observations)
+        top_patterns = pattern_counts.most_common(5)
+        summary[call_name] = {
+            "occurrences": len(observations),
+            "arg_count": {
+                "min": min(counts),
+                "max": max(counts),
+                "common": max(set(counts), key=counts.count),
+            },
+            "patterns": [
+                {"count": c, "args": n, "types": list(t)}
+                for (n, t), c in top_patterns
+            ],
+        }
+    return summary
+
 
 def _load_grammar(ext):
     entry = EXTENSION_MAP.get(ext)