fix: isolate API unit app imports from worker contracts

Harden worker contract eviction to drop deepest API-shadowed modules
first, and make job-poll unit tests reassert the API import root then
clear cached API app modules so shared pytest runs stay green.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: suguanYang

apps/api/tests/unit/test_job_poll_session_hygiene.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import sys
+from pathlib import Path
 from types import ModuleType, SimpleNamespace
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -18,9 +20,49 @@
 configure_import_environment()
 ensure_import_paths()
 
+_API_ROOT = str(Path(__file__).resolve().parents[2])
 
-def _load_api_modules() -> tuple[ModuleType, ModuleType, ModuleType, ModuleType]:
+
+def _prioritize_api_import_root() -> None:
+    """Keep apps/api ahead of apps/worker for the shared `app` package name."""
     ensure_import_paths()
+    if _API_ROOT in sys.path:
+        sys.path.remove(_API_ROOT)
+    sys.path.insert(0, _API_ROOT)
+
+
+def _is_api_app_module(module: ModuleType | None) -> bool:
+    if module is None:
+        return False
+    module_file = getattr(module, "__file__", None)
+    if isinstance(module_file, str) and module_file.startswith(_API_ROOT):
+        return True
+    module_paths = getattr(module, "__path__", ())
+    try:
+        return any(str(path).startswith(_API_ROOT) for path in module_paths)
+    except KeyError:
+        return False
+
+
+def _drop_non_api_app_modules() -> None:
+    for module_name in sorted(sys.modules, key=len, reverse=True):
+        if module_name != "app" and not module_name.startswith("app."):
+            continue
+        if not _is_api_app_module(sys.modules.get(module_name)):
+            sys.modules.pop(module_name, None)
+
+
+def _drop_api_app_modules() -> None:
+    for module_name in sorted(sys.modules, key=len, reverse=True):
+        if module_name != "app" and not module_name.startswith("app."):
+            continue
+        if _is_api_app_module(sys.modules.get(module_name)):
+            sys.modules.pop(module_name, None)
+
+
+def _load_api_modules() -> tuple[ModuleType, ModuleType, ModuleType, ModuleType]:
+    _prioritize_api_import_root()
+    _drop_non_api_app_modules()
     from app.services.auth import api_key_authentication_service
     from app.services.rate_limit import (
         data_structures,
@@ -36,6 +78,14 @@ def _load_api_modules() -> tuple[ModuleType, ModuleType, ModuleType, ModuleType]
     )
 
 
+@pytest.fixture(autouse=True)
+def _clear_api_app_modules_after_unit_test():
+    """Avoid leaving API's `app` package cached for later worker contract tests."""
+    yield
+    _drop_api_app_modules()
+
+
+
 @pytest.mark.asyncio
 async def test_get_tier_reuses_provided_session_without_get_db_context() -> None:
     _, _, _, tier_service = _load_api_modules()

apps/worker/tests/contract/conftest.py

@@ -37,7 +37,14 @@ def _module_loaded_from(module_name: str, root: Path) -> bool:
         return True
 
     module_paths = getattr(module, "__path__", ())
-    return any(str(module_path).startswith(root_value) for module_path in module_paths)
+    try:
+        return any(
+            str(module_path).startswith(root_value) for module_path in module_paths
+        )
+    except KeyError:
+        # Namespace path iteration can raise if a parent package was already
+        # removed from sys.modules mid-eviction.
+        return False
 
 
 def _ensure_worker_import_context() -> None:
@@ -46,11 +53,16 @@ def _ensure_worker_import_context() -> None:
         sys.path.remove(worker_root_value)
     sys.path.insert(0, worker_root_value)
 
-    cached_module_names = list(sys.modules)
-    for module_name in cached_module_names:
-        if module_name == "app" or module_name.startswith("app."):
-            if _module_loaded_from(module_name, _API_ROOT):
-                sys.modules.pop(module_name, None)
+    cached_module_names = [
+        module_name
+        for module_name in sys.modules
+        if module_name == "app" or module_name.startswith("app.")
+    ]
+    # Evict deepest modules first so namespace __path__ checks never observe a
+    # missing parent `app` entry while walking API-shadowed packages.
+    for module_name in sorted(cached_module_names, key=len, reverse=True):
+        if _module_loaded_from(module_name, _API_ROOT):
+            sys.modules.pop(module_name, None)
 
 
 @pytest.fixture(autouse=True)