feat(lint): any-on-boundary rules + any/exported metadata in TS parser

CHANGELOG.md

@@ -35,6 +35,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   only the identifier name appears in output, so a real secret never
   lands in CI logs or PR comments. The name match is singular-only
   (`tokens_in` token counts and `_FIX_TOKENS` word lists don't fire).
+- **Lint rule `db-call-in-loop` + `loop_depth` on call records.**
+  Both parsers now thread loop depth through their call-collection DFS;
+  CALLS edge metadata gains `loop_depth` / `in_loop`. TS additionally
+  treats function arguments to `.map/.forEach/.filter/.reduce/.flatMap`
+  as implicit loop bodies. The new rule (high) flags DB-API calls
+  (`db.select|insert|update|delete|execute|query` in TS;
+  `session|cursor|db . query|execute|add|commit` in Python) executing
+  inside a loop — the classic N+1.
+- **Risk-ranked untested + `new_untested_hotspot` review rule.**
+  `rank_untested()` joins untested × hotspot score × blast radius so
+  reports lead with untested *and* high-fan-in functions; surfaced in
+  `codegraph query untested` and the MCP `untested` tool. New default
+  review rule (high, threshold 5): a newly-added function with ≥5
+  callers and no test-module caller now gates PRs. **Note:** this new
+  default rule can change `codegraph review` exit codes for existing
+  users; disable by shipping a custom `.codegraph/rules.yml`.
+- **Lint rules `any-on-boundary` (low) and `any-into-db-write` (med).**
+  The TS parser records `exported`, `any_params`, and `any_return`
+  metadata; the rules flag exported functions with `any` in their
+  signature and `as any` casts passed into DB-write calls. Surface
+  syntax only — this is not type inference.
 
 ## [0.1.2] — 2026-05-31
 

codegraph/analysis/lint.py

@@ -24,7 +24,7 @@
 
 from codegraph.parsers.base import load_parser, node_text
 from codegraph.parsers.python import _is_test_file as _is_py_test_file
-from codegraph.parsers.typescript import EXT_TO_LANG
+from codegraph.parsers.typescript import _ANY_WORD_RE, EXT_TO_LANG
 from codegraph.parsers.typescript import _is_test_file as _is_ts_test_file
 
 if TYPE_CHECKING:
@@ -391,11 +391,155 @@ def _check_db_call_in_loop(
     )
 
 
+_DEFAULT_DB_WRITE_PATTERN = (
+    r"\b(db|database)\.(insert|update|execute)\b|\.values\(|\.set\("
+)
+
+_MAX_ANCESTOR_LEVELS = 6
+
+
+def _has_export_ancestor(node: tree_sitter.Node) -> bool:
+    """Return True if any ancestor (up to _MAX_ANCESTOR_LEVELS) is an export_statement."""
+    cur = node.parent
+    for _ in range(_MAX_ANCESTOR_LEVELS):
+        if cur is None:
+            return False
+        if cur.type == "export_statement":
+            return True
+        cur = cur.parent
+    return False
+
+
+def _as_any_in_args(args_node: tree_sitter.Node, src: bytes) -> bool:
+    """Return True if any argument inside *args_node* is an ``as any`` cast."""
+    for child in args_node.children:
+        if child.type == "as_expression":
+            right = child.child_by_field_name("type")
+            # tree-sitter represents the rhs of `as` as a 'predefined_type' child
+            # or via child_by_field_name("type"). Walk children to be safe.
+            for c in child.children:
+                if c.type == "predefined_type" and node_text(c, src) == "any":
+                    return True
+            if right is not None and node_text(right, src) == "any":
+                return True
+    return False
+
+
+def _check_any_on_boundary(
+    node: tree_sitter.Node, ctx: _LintContext
+) -> LintFinding | None:
+    """Exported function / arrow with ``any`` in a param type or return type."""
+    if ctx.language not in _TS_LANGS or ctx.is_test_file:
+        return None
+
+    # Match function declarations directly or inside export_statement
+    if node.type == "function_declaration":
+        if not _has_export_ancestor(node):
+            return None
+    elif node.type in ("arrow_function", "function_expression", "function"):
+        # Must be inside an export_statement (via lexical_declaration)
+        if not _has_export_ancestor(node):
+            return None
+    else:
+        return None
+
+    # Check params for any
+    params_node = node.child_by_field_name("parameters")
+    if params_node is None:
+        for c in node.children:
+            if c.type == "formal_parameters":
+                params_node = c
+                break
+
+    any_param_names: list[str] = []
+    if params_node is not None:
+        for param in params_node.children:
+            if param.type not in ("required_parameter", "optional_parameter"):
+                continue
+            ann = param.child_by_field_name("type")
+            if ann is not None:
+                type_text = node_text(ann, ctx.src)
+                if _ANY_WORD_RE.search(type_text):
+                    # Grab param name
+                    for c in param.children:
+                        if c.type == "identifier":
+                            any_param_names.append(node_text(c, ctx.src))
+                            break
+
+    # Check return type annotation
+    has_any_return = False
+    for c in node.children:
+        if c.type == "type_annotation":
+            if _ANY_WORD_RE.search(node_text(c, ctx.src)):
+                has_any_return = True
+            break
+
+    if not any_param_names and not has_any_return:
+        return None
+
+    parts: list[str] = []
+    if any_param_names:
+        parts.append(f"param(s) {', '.join(any_param_names)}")
+    if has_any_return:
+        parts.append("return type")
+    detail = " and ".join(parts)
+
+    # Find a name for the function (best-effort)
+    name_node = node.child_by_field_name("name")
+    func_name = node_text(name_node, ctx.src) if name_node is not None else "<anonymous>"
+
+    return LintFinding(
+        rule_id=ctx.rule.id,
+        severity=ctx.rule.severity,
+        message=_format_message(ctx.rule, name=func_name, detail=detail),
+        file=ctx.rel_path,
+        line=node.start_point[0] + 1,
+        snippet=_snippet(node, ctx.src),
+    )
+
+
+def _check_any_into_db_write(
+    node: tree_sitter.Node, ctx: _LintContext
+) -> LintFinding | None:
+    """``as any`` cast passed as argument to a DB-write call."""
+    if ctx.language not in _TS_LANGS or ctx.is_test_file:
+        return None
+    if node.type != "call_expression" or not _is_chain_root(node):
+        return None
+
+    chain = _member_chain(node, ctx.src)
+    if chain is None:
+        return None
+    base, methods = chain
+    chain_text = ".".join([base, *methods])
+    pattern = str(
+        ctx.rule.options.get("pattern") or _DEFAULT_DB_WRITE_PATTERN
+    )
+    if not re.search(pattern, chain_text):
+        return None
+
+    # Check whether any argument in the outermost call contains `as any`
+    args_node = node.child_by_field_name("arguments")
+    if args_node is None or not _as_any_in_args(args_node, ctx.src):
+        return None
+
+    return LintFinding(
+        rule_id=ctx.rule.id,
+        severity=ctx.rule.severity,
+        message=_format_message(ctx.rule, chain=chain_text),
+        file=ctx.rel_path,
+        line=node.start_point[0] + 1,
+        snippet=_snippet(node, ctx.src),
+    )
+
+
 _CHECK_REGISTRY: dict[str, _Checker] = {
     "console-in-prod": _check_console_in_prod,
     "unfiltered-query": _check_unfiltered_query,
     "sensitive-literal": _check_sensitive_literal,
     "db-call-in-loop": _check_db_call_in_loop,
+    "any-on-boundary": _check_any_on_boundary,
+    "any-into-db-write": _check_any_into_db_write,
 }
 
 
@@ -424,6 +568,18 @@ def _check_db_call_in_loop(
         severity="high",
         message="DB call `{chain}` inside a loop (N+1 risk)",
     ),
+    LintRule(
+        id="any-on-boundary",
+        check="any-on-boundary",
+        severity="low",
+        message="Exported function `{name}` uses `any` in {detail}",
+    ),
+    LintRule(
+        id="any-into-db-write",
+        check="any-into-db-write",
+        severity="med",
+        message="DB-write call `{chain}` receives an `as any` cast argument",
+    ),
 ]
 
 

codegraph/parsers/typescript.py

@@ -257,6 +257,20 @@ def _strip_type_annotation(text: str) -> str:
     return s.strip()
 
 
+_ANY_WORD_RE = re.compile(r"\bany\b")
+
+
+def _type_mentions_any(t: str | None) -> bool:
+    """Return True if *t* contains the bare type ``any``.
+
+    Matches ``any``, ``any[]``, ``Promise<any>``, ``Record<string, any>``
+    but NOT ``Company``, ``anything``, or similar sub-word occurrences.
+    """
+    if t is None:
+        return False
+    return bool(_ANY_WORD_RE.search(t))
+
+
 def _extract_param(
     p: tree_sitter.Node, src: bytes
 ) -> dict[str, str | None] | None:
@@ -682,13 +696,15 @@ def _visit(
                         self._handle_function_decl(
                             sub, rel, parent_qualname, parent_id, lang,
                             src, nodes, edges,
+                            exported=True,
                         )
                     elif sub.type in (
                         "lexical_declaration", "variable_declaration"
                     ):
                         self._handle_lexical_decl(
                             sub, rel, parent_qualname, parent_id, lang,
                             src, nodes, edges,
+                            exported=True,
                         )
 
     def _handle_import(
@@ -899,6 +915,8 @@ def _handle_function_decl(
         src: bytes,
         nodes: list[Node],
         edges: list[Edge],
+        *,
+        exported: bool = False,
     ) -> None:
         name_node = node.child_by_field_name("name")
         if name_node is None:
@@ -921,6 +939,17 @@ def _handle_function_decl(
             "params": params_list,
             "returns": return_type,
         }
+        if exported:
+            func_md["exported"] = True
+        any_params = [
+            p["name"]
+            for p in params_list
+            if isinstance(p, dict) and _type_mentions_any(p.get("type"))
+        ]
+        if any_params:
+            func_md["any_params"] = any_params
+        if _type_mentions_any(return_type):
+            func_md["any_return"] = True
         if _has_public_api_pragma_ts(node, src):
             func_md["public_api"] = True
         func_node = Node(
@@ -957,6 +986,8 @@ def _handle_lexical_decl(
         src: bytes,
         nodes: list[Node],
         edges: list[Edge],
+        *,
+        exported: bool = False,
     ) -> None:
         for child in node.children:
             if child.type != "variable_declarator":
@@ -990,6 +1021,23 @@ def _handle_lexical_decl(
                     value_node, arrow_params, src
                 )
 
+                arrow_md: dict[str, Any] = {
+                    "arrow": True,
+                    "params": params_list,
+                    "returns": return_type,
+                }
+                if exported:
+                    arrow_md["exported"] = True
+                any_params = [
+                    p["name"]
+                    for p in params_list
+                    if isinstance(p, dict) and _type_mentions_any(p.get("type"))
+                ]
+                if any_params:
+                    arrow_md["any_params"] = any_params
+                if _type_mentions_any(return_type):
+                    arrow_md["any_return"] = True
+
                 func_node = Node(
                     id=func_id,
                     kind=NodeKind.FUNCTION,
@@ -999,11 +1047,7 @@ def _handle_lexical_decl(
                     line_start=node.start_point[0] + 1,
                     line_end=node.end_point[0] + 1,
                     language=lang,
-                    metadata={
-                        "arrow": True,
-                        "params": params_list,
-                        "returns": return_type,
-                    },
+                    metadata=arrow_md,
                 )
                 nodes.append(func_node)
 

tests/fixtures/lint_sample/src/api.ts

@@ -0,0 +1,34 @@
+import { db, jobs } from "./schema";
+
+// Case A — exported function with any param and any return: SHOULD fire any-on-boundary
+export function processPayload(x: any): any {
+  return x;
+}
+
+// Case A — exported arrow with any param: SHOULD fire any-on-boundary
+export const handleInput = (data: any): string => {
+  return String(data);
+};
+
+// Typed exported counterexample — SHOULD NOT fire any-on-boundary
+export function typedHandler(x: string): string {
+  return x.trim();
+}
+
+// Non-exported function with any — SHOULD NOT fire any-on-boundary (case A)
+function internalProcess(x: any): any {
+  return x;
+}
+
+// Case B — as any into db.insert().values(): SHOULD fire any-into-db-write
+export async function submitJob(payload: unknown): Promise<void> {
+  await db.insert(jobs).values(payload as any);
+}
+
+// Typed call — SHOULD NOT fire any-into-db-write
+export async function submitTypedJob(payload: { name: string }): Promise<void> {
+  await db.insert(jobs).values(payload);
+}
+
+// Keep internalProcess reachable so tree-shaking doesn't flag it
+export { internalProcess };