learningcircuit--local-deep-research/.pre-commit-hooks/check-settings-manager-thread-safety.py

#!/usr/bin/env python3
"""
Pre-commit hook to detect get_settings_manager() calls without db_session
in code that runs in background threads.

Background threads have no Flask app context, so get_settings_manager()
without an explicit db_session falls back to JSON defaults only.  Settings
that have no JSON defaults (e.g. local_search_* embedding keys) silently
return None, causing user-configured values to be ignored.  See #3453.

Scope and limitations:
  - Only direct calls inside a thread function are inspected.  Indirect
    calls via a helper function (thread fn -> helper -> get_settings_manager)
    are not caught; that would require cross-function call-graph analysis.
  - Thread detection relies on the @thread_cleanup decorator or
    _background_/_auto_/*_worker naming conventions.  Other thread targets
    (e.g. threading.Thread(target=self._monitor_resources)) are not matched.
    Prefer decorating those functions with @thread_cleanup if they need
    to call get_settings_manager().
  - Test files are excluded via .pre-commit-config.yaml ``exclude: ^tests/``.
"""

import ast
import sys
from pathlib import Path
from typing import List, Tuple

# Decorators that mark a function as running in a background thread
THREAD_DECORATORS = frozenset({"thread_cleanup"})

# Function name patterns that indicate background-thread execution
THREAD_FUNCTION_PREFIXES = ("_background_", "_auto_")
THREAD_FUNCTION_SUFFIXES = ("_worker",)


class SettingsManagerThreadSafetyChecker(ast.NodeVisitor):
    """AST visitor to detect unsafe get_settings_manager() calls in thread code."""

    def __init__(self, filename: str):
        self.filename = filename
        self.issues: List[Tuple[int, str]] = []

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        if self._is_thread_function(node):
            self._check_body_for_unsafe_calls(node)
        self.generic_visit(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        if self._is_thread_function(node):
            self._check_body_for_unsafe_calls(node)
        self.generic_visit(node)

    # ------------------------------------------------------------------

    def _is_thread_function(self, node) -> bool:
        """Return True if the function is likely executed in a background thread."""
        # Check decorators
        for decorator in node.decorator_list:
            name = None
            if isinstance(decorator, ast.Name):
                name = decorator.id
            elif isinstance(decorator, ast.Attribute):
                name = decorator.attr
            elif isinstance(decorator, ast.Call):
                if isinstance(decorator.func, ast.Name):
                    name = decorator.func.id
                elif isinstance(decorator.func, ast.Attribute):
                    name = decorator.func.attr
            if name and name in THREAD_DECORATORS:
                return True

        # Check function name conventions
        fname = node.name
        if any(fname.startswith(p) for p in THREAD_FUNCTION_PREFIXES):
            return True
        if any(fname.endswith(s) for s in THREAD_FUNCTION_SUFFIXES):
            return True

        return False

    def _check_body_for_unsafe_calls(self, node) -> None:
        """Walk the function body looking for get_settings_manager() without db_session.

        Stops at nested function-def boundaries: any nested function is
        visited separately by ``generic_visit`` -> ``visit_FunctionDef``,
        so we must not descend into it here or we would double-report.
        """
        for child in self._iter_non_nested(node):
            if not isinstance(child, ast.Call):
                continue
            if not self._is_get_settings_manager_call(child):
                continue
            if not self._has_db_session_arg(child):
                self.issues.append(
                    (
                        child.lineno,
                        "get_settings_manager() called without db_session= "
                        "in a background-thread function. In threads without "
                        "Flask app context the DB session will be None and "
                        "settings fall back to JSON defaults only. "
                        "Pass an explicit db_session from get_user_db_session().",
                    ),
                )

    @classmethod
    def _iter_non_nested(cls, node):
        """Yield descendants of ``node``, stopping at nested function defs."""
        for child in ast.iter_child_nodes(node):
            if isinstance(
                child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)
            ):
                continue
            yield child
            yield from cls._iter_non_nested(child)

    @staticmethod
    def _is_get_settings_manager_call(call_node: ast.Call) -> bool:
        func = call_node.func
        if isinstance(func, ast.Name) and func.id == "get_settings_manager":
            return True
        if (
            isinstance(func, ast.Attribute)
            and func.attr == "get_settings_manager"
        ):
            return True
        return False

    @staticmethod
    def _has_db_session_arg(call_node: ast.Call) -> bool:
        # db_session is the first positional param of get_settings_manager,
        # so any positional argument satisfies the safety contract.
        if call_node.args:
            return True
        return any(kw.arg == "db_session" for kw in call_node.keywords)


def check_file(filepath: Path) -> List[Tuple[str, int, str]]:
    """Check a single Python file."""
    issues = []
    try:
        content = filepath.read_text(encoding="utf-8")
        tree = ast.parse(content, filename=str(filepath))
        checker = SettingsManagerThreadSafetyChecker(str(filepath))
        checker.visit(tree)
        for line_no, message in checker.issues:
            issues.append((str(filepath), line_no, message))
    except SyntaxError as e:
        print(f"Syntax error in {filepath}: {e}", file=sys.stderr)
    except Exception as e:
        print(f"Error checking {filepath}: {e}", file=sys.stderr)
    return issues


def main():
    files_to_check = sys.argv[1:] if len(sys.argv) > 1 else []
    if not files_to_check:
        print("No files to check")
        return 0

    all_issues = []
    for filepath_str in files_to_check:
        filepath = Path(filepath_str)
        if not filepath.suffix == ".py":
            continue
        issues = check_file(filepath)
        all_issues.extend(issues)

    if all_issues:
        print(
            "\n\u274c get_settings_manager() called without db_session "
            "in background-thread code:\n"
        )
        for filepath, line_no, message in all_issues:
            print(f"  {filepath}:{line_no}: {message}")
        print(
            "\n\U0001f4a1 Tip: Use get_user_db_session() to obtain a session "
            "and pass it explicitly:\n"
        )
        print(
            "  with get_user_db_session(username, db_password) as db_session:"
        )
        print(
            "      settings = get_settings_manager("
            "db_session=db_session, username=username)"
        )
        print()
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())