Add AST function-source helper import boundary guard

Co-authored-by: Shri Sukhani <shrisukhani@users.noreply.github.com>

Author: cursoragent

CONTRIBUTING.md

@@ -89,6 +89,7 @@ This runs lint, format checks, compile checks, tests, and package build.
   - `tests/test_agent_terminal_status_helper_usage.py` (shared agent terminal-status helper usage enforcement),
   - `tests/test_architecture_marker_usage.py` (architecture marker coverage across guard modules),
   - `tests/test_ast_function_source_helper_usage.py` (shared AST function-source helper usage enforcement across architecture guard suites),
+  - `tests/test_ast_function_source_import_boundary.py` (shared AST function-source helper import boundary enforcement across test modules),
   - `tests/test_ast_function_source_utils.py` (shared AST function-source helper contract validation),
   - `tests/test_binary_file_open_helper_usage.py` (shared binary file open helper usage enforcement),
   - `tests/test_browser_use_payload_helper_usage.py` (browser-use payload helper usage enforcement),

tests/test_architecture_marker_usage.py

@@ -17,6 +17,7 @@
     "tests/test_agent_task_read_helper_usage.py",
     "tests/test_agent_stop_helper_usage.py",
     "tests/test_agent_terminal_status_helper_usage.py",
+    "tests/test_ast_function_source_import_boundary.py",
     "tests/test_ast_function_source_helper_usage.py",
     "tests/test_ast_function_source_utils.py",
     "tests/test_guardrail_ast_utils.py",

tests/test_ast_function_source_helper_usage.py

@@ -1,3 +1,4 @@
+import ast
 from pathlib import Path
 
 import pytest
@@ -21,12 +22,23 @@
     "tests/test_web_request_wrapper_internal_reuse.py",
 )
 
+def _imports_collect_function_sources(module_text: str) -> bool:
+    module_ast = ast.parse(module_text)
+    for node in module_ast.body:
+        if not isinstance(node, ast.ImportFrom):
+            continue
+        if node.module != "tests.ast_function_source_utils":
+            continue
+        if any(alias.name == "collect_function_sources" for alias in node.names):
+            return True
+    return False
+
 
 def test_ast_guard_modules_reuse_shared_collect_function_sources_helper():
     violating_modules: list[str] = []
     for module_path in AST_FUNCTION_SOURCE_GUARD_MODULES:
         module_text = Path(module_path).read_text(encoding="utf-8")
-        if "ast_function_source_utils import collect_function_sources" not in module_text:
+        if not _imports_collect_function_sources(module_text):
             violating_modules.append(module_path)
             continue
         if "collect_function_sources(" not in module_text:
@@ -49,7 +61,7 @@ def test_ast_guard_inventory_stays_in_sync_with_helper_imports():
         if normalized_path in excluded_modules:
             continue
         module_text = module_path.read_text(encoding="utf-8")
-        if "ast_function_source_utils import collect_function_sources" not in module_text:
+        if not _imports_collect_function_sources(module_text):
             continue
         discovered_modules.append(normalized_path)
 

tests/test_ast_function_source_import_boundary.py

@@ -0,0 +1,46 @@
+import ast
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.architecture
+
+
+EXPECTED_AST_FUNCTION_SOURCE_IMPORTER_MODULES = (
+    "tests/test_agent_request_wrapper_internal_reuse.py",
+    "tests/test_ast_function_source_utils.py",
+    "tests/test_extension_request_function_parse_boundary.py",
+    "tests/test_job_request_route_builder_internal_reuse.py",
+    "tests/test_job_request_wrapper_internal_reuse.py",
+    "tests/test_model_request_function_parse_boundary.py",
+    "tests/test_model_request_function_transport_boundary.py",
+    "tests/test_model_request_wrapper_internal_reuse.py",
+    "tests/test_request_wrapper_internal_reuse.py",
+    "tests/test_session_recordings_follow_redirects_boundary.py",
+    "tests/test_session_request_function_parse_boundary.py",
+    "tests/test_session_request_wrapper_internal_reuse.py",
+    "tests/test_session_resource_wrapper_internal_reuse.py",
+    "tests/test_web_request_wrapper_internal_reuse.py",
+)
+
+def _imports_collect_function_sources(module_text: str) -> bool:
+    module_ast = ast.parse(module_text)
+    for node in module_ast.body:
+        if not isinstance(node, ast.ImportFrom):
+            continue
+        if node.module != "tests.ast_function_source_utils":
+            continue
+        if any(alias.name == "collect_function_sources" for alias in node.names):
+            return True
+    return False
+
+
+def test_ast_function_source_helper_imports_are_centralized():
+    discovered_modules: list[str] = []
+    for module_path in sorted(Path("tests").glob("test_*.py")):
+        module_text = module_path.read_text(encoding="utf-8")
+        if not _imports_collect_function_sources(module_text):
+            continue
+        discovered_modules.append(module_path.as_posix())
+
+    assert discovered_modules == list(EXPECTED_AST_FUNCTION_SOURCE_IMPORTER_MODULES)