diff --git a/api/tests/conftest.py b/api/tests/conftest.py
index 194211f1af0e25..09d1ffe352e6db 100644
--- a/api/tests/conftest.py
+++ b/api/tests/conftest.py
@@ -1,3 +1,4 @@
+import gc
 import os
 from collections.abc import Iterator
 
@@ -10,6 +11,17 @@
 os.environ.setdefault("HUMAN_INPUT_IM_EVENT_TRANSPORT_MODE", "WEBHOOK")
 
 
+def pytest_collection_finish(session: pytest.Session) -> None:
+    """Move everything loaded during collection out of the garbage collector's reach.
+
+    Collection imports the whole application, so every automatic full collection afterwards
+    rescanned those long-lived objects and stalled whichever test triggered it. Objects created
+    by tests themselves are still collected normally.
+    """
+    gc.collect()
+    gc.freeze()
+
+
 @pytest.fixture(autouse=True)
 def _bind_workflow_file_runtime() -> Iterator[None]:
     with use_workflow_file_runtime(create_dify_workflow_file_runtime()):
diff --git a/api/tests/e2e_tests/contracts/controllers/web/test_message_list.py b/api/tests/e2e_tests/contracts/controllers/web/test_message_list.py
index 4542a5afbab369..4c9a499451c9cc 100644
--- a/api/tests/e2e_tests/contracts/controllers/web/test_message_list.py
+++ b/api/tests/e2e_tests/contracts/controllers/web/test_message_list.py
@@ -56,7 +56,7 @@ def do_HEAD(self) -> None:
     http_pool = http_client_pooling.HttpClientPoolFactory()
     monkeypatch.setattr(http_client_pooling, "_factory", http_pool)
     server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-    thread = Thread(target=server.serve_forever, daemon=True)
+    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     thread.start()
     try:
         yield f"http://127.0.0.1:{server.server_port}", requests
diff --git a/api/tests/helpers/unit/services/conversation_name_fixtures.py b/api/tests/helpers/unit/services/conversation_name_fixtures.py
index be8188e4fe1e44..2295f873f6b3a1 100644
--- a/api/tests/helpers/unit/services/conversation_name_fixtures.py
+++ b/api/tests/helpers/unit/services/conversation_name_fixtures.py
@@ -95,7 +95,7 @@ def respond(self, data: object, *, stream: bool, code: int = 0, message: str = "
             self.wfile.write(body)
 
     server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-    thread = Thread(target=server.serve_forever, daemon=True)
+    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     original_url = plugin_http.plugin_daemon_inner_api_baseurl
     # Change real connection settings, preserving every production client and method.
     plugin_http.plugin_daemon_inner_api_baseurl = URL(f"http://127.0.0.1:{server.server_port}")
diff --git a/api/tests/unit_tests/conftest.py b/api/tests/unit_tests/conftest.py
index cbcee1bf7d6e0c..9cfe7db5de22c9 100644
--- a/api/tests/unit_tests/conftest.py
+++ b/api/tests/unit_tests/conftest.py
@@ -5,8 +5,10 @@
 from collections.abc import Callable, Iterator
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from threading import Lock
 from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, create_autospec, patch
+from weakref import WeakKeyDictionary
 
 # Ensure OpenDAL fs writes to tmp to avoid polluting workspace
 os.environ.setdefault("OPENDAL_SCHEME", "fs")
@@ -17,7 +19,7 @@
 from dify_trace_sdk.testing import RecordingEgress
 from flask import Flask
 from redis import Redis
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event
 from sqlalchemy.engine import URL, Engine
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -132,7 +134,10 @@ def _patch_redis_clients() -> Iterator[None]:
 @pytest.fixture(autouse=True)
 def reset_redis_mock(_patch_redis_clients: None) -> None:
     """Reset the shared Redis mock after per-test client rebinding."""
-    redis_mock.reset_mock()
+    # ``mock_calls`` records calls on every attached child and return value, so when it is empty the
+    # recursive reset has nothing to clear. Most tests never touch Redis; skip walking ~370 child mocks.
+    if redis_mock.mock_calls:
+        redis_mock.reset_mock()
     # Restoring a monkeypatched method can leave it detached from the parent's reset traversal.
     redis_mock.delete.reset_mock()
     redis_mock.get.reset_mock()
@@ -178,6 +183,21 @@ def apply(**values: object) -> None:
     return apply
 
 
+# The callable that materializes each live ``_sqlite_engine``'s database file.
+_sqlite_copiers: WeakKeyDictionary[Engine, Callable[[], None]] = WeakKeyDictionary()
+
+
+def _copy_sqlite_template(template: Path, database_path: Path, lock: Lock) -> None:
+    """Create ``database_path`` as a copy of ``template`` unless it already exists.
+
+    Idempotent and safe across threads sharing ``lock``: the copy happens at most once, so an
+    open connection never sees its file overwritten.
+    """
+    with lock:
+        if not database_path.exists():
+            shutil.copyfile(template, database_path)
+
+
 @pytest.fixture
 def _sqlite_engine(_sqlite_database_template: Path) -> Iterator[Engine]:
     """Copy the schema into an isolated directory without pytest's numbered scan.
@@ -185,12 +205,21 @@ def _sqlite_engine(_sqlite_database_template: Path) -> Iterator[Engine]:
     ``tmp_path`` searches all preceding test directories for a free number. This
     autouse dependency needs only a unique, disposable directory, including any
     SQLite journal files, so keep it separate from test-owned ``tmp_path`` data.
+
+    Most tests never connect, so the schema is copied on the engine's first connection
+    rather than for every test. Code that opens the file without this engine must go
+    through ``sqlite_engine``, which copies it up front.
     """
     with TemporaryDirectory(prefix="case-", dir=_sqlite_database_template.parent) as directory:
         database_path = Path(directory) / "unit-tests.sqlite3"
-        shutil.copyfile(_sqlite_database_template, database_path)
         engine = create_engine(URL.create("sqlite", database=str(database_path)))
+        lock = Lock()
+
+        def copy() -> None:
+            _copy_sqlite_template(_sqlite_database_template, database_path, lock)
 
+        event.listen(engine, "do_connect", lambda *_args: copy())
+        _sqlite_copiers[engine] = copy
         try:
             yield engine
         finally:
@@ -236,8 +265,13 @@ def _unbound_session_factory(
 
 @pytest.fixture
 def sqlite_engine(_sqlite_engine: Engine) -> Engine:
-    """Expose the pristine full-schema SQLite engine to tests."""
+    """Expose the pristine full-schema SQLite engine to tests.
+
+    The database file exists on return, so tests may hand ``sqlite_engine.url`` to code
+    that opens it with an engine of its own.
+    """
 
+    _sqlite_copiers[_sqlite_engine]()
     return _sqlite_engine
 
 
diff --git a/api/tests/unit_tests/controllers/console/datasets/test_data_source.py b/api/tests/unit_tests/controllers/console/datasets/test_data_source.py
index 2f62faf8b7d44a..77ec458bf22e26 100644
--- a/api/tests/unit_tests/controllers/console/datasets/test_data_source.py
+++ b/api/tests/unit_tests/controllers/console/datasets/test_data_source.py
@@ -320,10 +320,25 @@ class AdmittedAccount:
     status: str
 
 
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the console namespace, which takes a few hundred milliseconds to mount.
+
+    Each test registers its own ``extensions["application_services"]`` for the duration of that test.
+    """
+    app = Flask(__name__)
+    app.config.update(TESTING=False, PROPAGATE_EXCEPTIONS=False)
+    api = ExternalApi(app)
+    api.error_handlers = console_api.error_handlers.copy()
+    api.add_namespace(console_ns, path="/console/api")
+    return app
+
+
 class TestDataSourceCases:
     @staticmethod
     @pytest.fixture
     def collaborators(
+        console_app: Flask,
         config_overrides: Callable[..., None],
         monkeypatch: pytest.MonkeyPatch,
     ) -> ControllerCollaborators:
@@ -333,12 +348,6 @@ def collaborators(
             RBAC_ENABLED=False,
             CONSOLE_API_URL="https://console-api.example.com",
         )
-        app = Flask(__name__)
-        app.config.update(TESTING=False, PROPAGATE_EXCEPTIONS=False)
-        api = ExternalApi(app)
-        api.error_handlers = console_api.error_handlers.copy()
-        api.add_namespace(console_ns, path="/console/api")
-
         account = AdmittedAccount(id=ACCOUNT_ID, role="owner", status="active")
         account_with_tenant = AccountWithTenant(account=cast("Account", account), tenant_id=TENANT_ID)
         monkeypatch.setattr(
@@ -393,7 +402,7 @@ def collaborators(
         )
         estimates = RecordingEstimateService()
         dispatcher = RecordingDispatcher()
-        app.extensions["application_services"] = ApplicationServiceRegistry(
+        services = ApplicationServiceRegistry(
             data_sources=DataSourceApplicationServices(
                 bindings=DataSourceBindingApplicationService(bindings=bindings),
                 notion_imports=NotionImportApplicationService(
@@ -412,8 +421,9 @@ def collaborators(
                 indexing_estimates=estimates,
             ),
         )
+        monkeypatch.setitem(console_app.extensions, "application_services", services)
         return ControllerCollaborators(
-            client=app.test_client(),
+            client=console_app.test_client(),
             account=account,
             bindings=bindings,
             notion_source=notion_source,
diff --git a/api/tests/unit_tests/controllers/console/human_input_v2/test_contact.py b/api/tests/unit_tests/controllers/console/human_input_v2/test_contact.py
index 4be06966e5037c..25f89e1b25376c 100644
--- a/api/tests/unit_tests/controllers/console/human_input_v2/test_contact.py
+++ b/api/tests/unit_tests/controllers/console/human_input_v2/test_contact.py
@@ -50,21 +50,32 @@ def current_account(sqlite_session_factory: sessionmaker[Session]) -> Account:
     return account
 
 
-@pytest.fixture(params=[False, True], ids=["legacy-roles", "rbac"])
-def client(current_account: Account, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> FlaskClient:
-    monkeypatch.setattr(dify_config, "DEPLOYMENT_EDITION", DeploymentEdition.CLOUD)
-    monkeypatch.setattr(dify_config, "LOGIN_DISABLED", True)
-    monkeypatch.setattr(dify_config, "RBAC_ENABLED", request.param)
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the whole console blueprint, which takes a few hundred milliseconds to register.
 
+    Each request logs in the account the running test stored under ``extensions["test_login_user"]``.
+    """
     app = Flask(__name__)
     LoginManager().init_app(app)
     app.register_blueprint(console_bp)
 
     @app.before_request
     def authenticate() -> None:
-        g._login_user = current_account
+        g._login_user = app.extensions["test_login_user"]
+
+    return app
 
-    return app.test_client()
+
+@pytest.fixture(params=[False, True], ids=["legacy-roles", "rbac"])
+def client(
+    console_app: Flask, current_account: Account, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> FlaskClient:
+    monkeypatch.setattr(dify_config, "DEPLOYMENT_EDITION", DeploymentEdition.CLOUD)
+    monkeypatch.setattr(dify_config, "LOGIN_DISABLED", True)
+    monkeypatch.setattr(dify_config, "RBAC_ENABLED", request.param)
+    monkeypatch.setitem(console_app.extensions, "test_login_user", current_account)
+    return console_app.test_client()
 
 
 @pytest.mark.parametrize(
diff --git a/api/tests/unit_tests/controllers/console/test_human_input_channel_wire.py b/api/tests/unit_tests/controllers/console/test_human_input_channel_wire.py
index 5700bb0d38306a..45d821e939fc21 100644
--- a/api/tests/unit_tests/controllers/console/test_human_input_channel_wire.py
+++ b/api/tests/unit_tests/controllers/console/test_human_input_channel_wire.py
@@ -50,8 +50,19 @@
 }
 
 
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the whole console blueprint, which takes a few hundred milliseconds to register."""
+    application = Flask(__name__)
+    LoginManager().init_app(application)
+    application.config["TESTING"] = False
+    application.config["RESTX_ERROR_404_HELP"] = False
+    application.register_blueprint(console_bp)
+    return application
+
+
 @pytest.fixture
-def wire_app(monkeypatch: pytest.MonkeyPatch) -> Flask:
+def wire_app(console_app: Flask, monkeypatch: pytest.MonkeyPatch) -> Flask:
     account = Account(name="Owner", email="owner@example.com", status=AccountStatus.ACTIVE)
     account.id = "account-1"
     account.role = TenantAccountRole.OWNER
@@ -64,13 +75,7 @@ def wire_app(monkeypatch: pytest.MonkeyPatch) -> Flask:
         "controllers.console.wraps.current_account_with_tenant",
         lambda: AccountWithTenant(account=account, tenant_id="workspace-1"),
     )
-
-    application = Flask(__name__)
-    LoginManager().init_app(application)
-    application.config["TESTING"] = False
-    application.config["RESTX_ERROR_404_HELP"] = False
-    application.register_blueprint(console_bp)
-    return application
+    return console_app
 
 
 def test_console_blueprint_registers_only_the_canonical_channel_routes(wire_app: Flask) -> None:
diff --git a/api/tests/unit_tests/controllers/console/test_human_input_v2_form_read.py b/api/tests/unit_tests/controllers/console/test_human_input_v2_form_read.py
index cd3175b225f49c..84f2f6ec9a4f41 100644
--- a/api/tests/unit_tests/controllers/console/test_human_input_v2_form_read.py
+++ b/api/tests/unit_tests/controllers/console/test_human_input_v2_form_read.py
@@ -56,8 +56,34 @@ class ConsoleForm:
     actor: Account | None
 
 
+@pytest.fixture(scope="module")
+def console_app() -> DifyApp:
+    """One app with the console namespace, which takes a few hundred milliseconds to mount.
+
+    Each request logs in the actor of the scenario the running test stored under
+    ``extensions["test_console_form"]``, read per request so tests can change it.
+    """
+    app = DifyApp(__name__)
+    manager = DifyLoginManager(app)
+
+    @manager.unauthorized_handler
+    def unauthorized() -> Response:
+        return Response(
+            '{"code":"unauthorized","message":"Login required","status":401}', status=401, mimetype="application/json"
+        )
+
+    @app.before_request
+    def load_session_actor() -> None:
+        g._login_user = app.extensions["test_console_form"].actor
+
+    ExternalApi(app).add_namespace(controller.console_ns, path="/console/api")
+    return app
+
+
 @pytest.fixture
-def console_form(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsoleForm:
+def console_form(
+    console_app: DifyApp, sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch
+) -> ConsoleForm:
     tenant = make_tenant(tenant_id=str(uuid4()))
     active_tenant = make_tenant(tenant_id=str(uuid4()))
     account = make_account(account_id=str(uuid4()), tenant=active_tenant)
@@ -113,22 +139,10 @@ def console_form(sqlite_session_factory: sessionmaker[Session], monkeypatch: pyt
     monkeypatch.setattr("services.human_input_v2.submission_service.naive_utc_now", lambda: _NOW)
     # Isolate session decoding and CSRF; keep the route's login and Contact authorization real.
     monkeypatch.setattr("libs.login.check_csrf_token", lambda _request, _user_id: None)
-    app = DifyApp(__name__)
-    manager = DifyLoginManager(app)
-
-    @manager.unauthorized_handler
-    def unauthorized() -> Response:
-        return Response(
-            '{"code":"unauthorized","message":"Login required","status":401}', status=401, mimetype="application/json"
-        )
-
-    scenario = ConsoleForm(app.test_client(), sqlite_session_factory, form, recipient, ContactId(identity.id), account)
-
-    @app.before_request
-    def load_session_actor() -> None:
-        g._login_user = scenario.actor
-
-    ExternalApi(app).add_namespace(controller.console_ns, path="/console/api")
+    scenario = ConsoleForm(
+        console_app.test_client(), sqlite_session_factory, form, recipient, ContactId(identity.id), account
+    )
+    monkeypatch.setitem(console_app.extensions, "test_console_form", scenario)
     return scenario
 
 
diff --git a/api/tests/unit_tests/controllers/service_api/app/test_human_input_form_compatibility.py b/api/tests/unit_tests/controllers/service_api/app/test_human_input_form_compatibility.py
index 3180bff3468e6d..18f172385297f0 100644
--- a/api/tests/unit_tests/controllers/service_api/app/test_human_input_form_compatibility.py
+++ b/api/tests/unit_tests/controllers/service_api/app/test_human_input_form_compatibility.py
@@ -100,8 +100,21 @@ def submission(self) -> dict[str, JsonValue] | None:
             return json.loads(form.raw_submission_inputs) if form.raw_submission_inputs is not None else None
 
 
+@pytest.fixture(scope="module")
+def service_app() -> Flask:
+    """One app with the service API namespace, which takes a few hundred milliseconds to mount.
+
+    Each test binds its own SQLite engine to the app for the duration of that test.
+    """
+    application_http = Flask(__name__)
+    LoginManager().init_app(application_http)
+    ExternalApi(application_http, register_default_root=False).add_namespace(controller.service_api_ns, path="/v1")
+    return application_http
+
+
 @pytest.fixture(params=["1", "2"])
 def approval(
+    service_app: Flask,
     sqlite_session_factory: sessionmaker[Session],
     sqlite_engine: Engine,
     monkeypatch: pytest.MonkeyPatch,
@@ -270,12 +283,9 @@ def publish(*, kwargs: dict[str, object]) -> None:
         published.append(kwargs)
 
     monkeypatch.setattr(resume_app_execution, "apply_async", publish)
-    application_http = Flask(__name__)
-    database._app_engines[application_http] = {None: sqlite_engine}
-    LoginManager().init_app(application_http)
-    ExternalApi(application_http, register_default_root=False).add_namespace(controller.service_api_ns, path="/v1")
+    monkeypatch.setitem(database._app_engines, service_app, {None: sqlite_engine})
     yield Approval(
-        application_http.test_client(),
+        service_app.test_client(),
         sqlite_session_factory,
         version,
         tenant.id,
diff --git a/api/tests/unit_tests/core/human_input_v2/im_provider/test_feishu_lark_sdk_boundary.py b/api/tests/unit_tests/core/human_input_v2/im_provider/test_feishu_lark_sdk_boundary.py
index a6cb5431680bd6..6d336299bfa7f5 100644
--- a/api/tests/unit_tests/core/human_input_v2/im_provider/test_feishu_lark_sdk_boundary.py
+++ b/api/tests/unit_tests/core/human_input_v2/im_provider/test_feishu_lark_sdk_boundary.py
@@ -588,7 +588,7 @@ def sdk_server(monkeypatch: pytest.MonkeyPatch) -> Generator[_RunningServer]:
     state = _ServerState()
     server = _SDKHTTPServer(("127.0.0.1", 0), _SDKRequestHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     thread.start()
     try:
         host, port = server.server_address[:2]
diff --git a/api/tests/unit_tests/core/human_input_v2/im_provider/test_ms_teams_sdk_boundary.py b/api/tests/unit_tests/core/human_input_v2/im_provider/test_ms_teams_sdk_boundary.py
index 3d20a34b97b49e..c7f85c1441beb9 100644
--- a/api/tests/unit_tests/core/human_input_v2/im_provider/test_ms_teams_sdk_boundary.py
+++ b/api/tests/unit_tests/core/human_input_v2/im_provider/test_ms_teams_sdk_boundary.py
@@ -184,7 +184,7 @@ def connector_server() -> Generator[_ConnectorServer]:
     state = _ConnectorState()
     server = _ConnectorHTTPServer(("127.0.0.1", 0), _ConnectorHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
     thread.start()
     try:
         yield _ConnectorServer(server, state)
diff --git a/api/tests/unit_tests/core/human_input_v2/im_provider/test_slack_sdk_boundary.py b/api/tests/unit_tests/core/human_input_v2/im_provider/test_slack_sdk_boundary.py
index cfe315748f4f7e..d215cf97fa92b7 100644
--- a/api/tests/unit_tests/core/human_input_v2/im_provider/test_slack_sdk_boundary.py
+++ b/api/tests/unit_tests/core/human_input_v2/im_provider/test_slack_sdk_boundary.py
@@ -161,7 +161,7 @@ def slack_api_server() -> Generator[_SlackApiServer]:
     state = _SlackApiState()
     server = _SlackApiHTTPServer(("127.0.0.1", 0), _SlackApiHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
     thread.start()
     try:
         yield _SlackApiServer(server, state)
diff --git a/api/tests/unit_tests/dev/test_generate_swagger_specs.py b/api/tests/unit_tests/dev/test_generate_swagger_specs.py
index 5d2ead032a79e4..a66ee9365f66ca 100644
--- a/api/tests/unit_tests/dev/test_generate_swagger_specs.py
+++ b/api/tests/unit_tests/dev/test_generate_swagger_specs.py
@@ -72,6 +72,16 @@ def _reset_schema_cache(api):
     api.__dict__.pop("__schema__", None)
 
 
+@pytest.fixture(scope="module")
+def generated_specs(tmp_path_factory: pytest.TempPathFactory) -> list[Path]:
+    """Spec files written once by ``generate_specs`` for the tests that only read them.
+
+    Each export imports every API surface and takes seconds, so read-only checks share one;
+    tests that exercise generation itself still call ``generate_specs`` on their own.
+    """
+    return _load_generate_swagger_specs_module().generate_specs(tmp_path_factory.mktemp("specs"))
+
+
 @pytest.mark.timeout(150)  # Two cold, coverage-instrumented CLI exports each have a 60-second limit.
 def test_console_stdout_export_contains_both_surfaces_without_spec_files(tmp_path: Path):
     script = Path(__file__).resolve().parents[3] / "dev" / "export_console_openapi.py"
@@ -105,12 +115,8 @@ def test_console_stdout_export_contains_both_surfaces_without_spec_files(tmp_pat
     assert not list(tmp_path.rglob("*.json"))
 
 
-def test_generate_specs_writes_all_openapi_files(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-
-    assert [path.name for path in written_paths] == [
+def test_generate_specs_writes_all_openapi_files(generated_specs: list[Path]):
+    assert [path.name for path in generated_specs] == [
         "console-openapi.json",
         "web-openapi.json",
         "service-openapi.json",
@@ -118,12 +124,12 @@ def test_generate_specs_writes_all_openapi_files(tmp_path: Path):
         "inner-openapi.json",
     ]
 
-    for path in written_paths:
+    for path in generated_specs:
         payload = json.loads(path.read_text(encoding="utf-8"))
         assert payload["openapi"] == "3.1.0"
         assert "paths" in payload
 
-    inner_path = next(path for path in written_paths if path.name == "inner-openapi.json")
+    inner_path = next(path for path in generated_specs if path.name == "inner-openapi.json")
     inner_payload = json.loads(inner_path.read_text(encoding="utf-8"))
     assert inner_payload["security"] == [{"InnerApiKey": []}]
     assert inner_payload["components"]["securitySchemes"] == {
@@ -229,12 +235,8 @@ def test_apply_runtime_defaults_forces_swagger_routes_on(monkeypatch):
     assert dify_config.SWAGGER_UI_ENABLED is True
 
 
-def test_generate_specs_writes_openapi_with_resolvable_references_and_null_defaults(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-
-    for path in written_paths:
+def test_generate_specs_writes_openapi_with_resolvable_references_and_null_defaults(generated_specs: list[Path]):
+    for path in generated_specs:
         payload = json.loads(path.read_text(encoding="utf-8"))
         schemas = payload["components"]["schemas"]
         refs = {
@@ -248,7 +250,7 @@ def test_generate_specs_writes_openapi_with_resolvable_references_and_null_defau
         assert refs <= set(schemas)
         assert all("nullable" not in value for value in _walk_values(payload) if isinstance(value, dict))
 
-    service_payload = json.loads((tmp_path / "service-openapi.json").read_text(encoding="utf-8"))
+    service_payload = json.loads((generated_specs[0].parent / "service-openapi.json").read_text(encoding="utf-8"))
     conversation_id = service_payload["components"]["schemas"]["ChatRequestPayload"]["properties"]["conversation_id"]
     assert "default" in conversation_id
     assert conversation_id["default"] is None
@@ -267,12 +269,8 @@ def test_generate_specs_writes_openapi_with_resolvable_references_and_null_defau
     assert document_detail["properties"]["tokens"]["default"] is None
 
 
-def test_generate_specs_writes_unique_operation_ids(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-
-    for path in written_paths:
+def test_generate_specs_writes_unique_operation_ids(generated_specs: list[Path]):
+    for path in generated_specs:
         payload = json.loads(path.read_text(encoding="utf-8"))
         operation_ids = list(_operation_ids(payload))
 
@@ -337,10 +335,8 @@ def test_finalize_openapi_payload_only_marks_explicit_binary_responses():
     assert "x-dify-binary-response-media-types" not in operation
 
 
-def test_console_file_and_audio_responses_declare_binary_success_and_json_errors(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-    module.generate_specs(tmp_path)
-    payload = json.loads((tmp_path / "console-openapi.json").read_text(encoding="utf-8"))
+def test_console_file_and_audio_responses_declare_binary_success_and_json_errors(generated_specs: list[Path]):
+    payload = json.loads((generated_specs[0].parent / "console-openapi.json").read_text(encoding="utf-8"))
     audio_media = {"audio/aac", "audio/flac", "audio/mp4", "audio/mpeg", "audio/ogg", "audio/wav", "audio/webm"}
     operations = [
         ("/workspaces/current/plugin/asset", "get", {"application/octet-stream"}),
@@ -360,10 +356,7 @@ def test_console_file_and_audio_responses_declare_binary_success_and_json_errors
             assert set(response["content"]) == {"application/json"}, (path, status)
 
 
-def test_system_features_specs_exclude_backend_only_fields(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
+def test_system_features_specs_exclude_backend_only_fields(generated_specs: list[Path]):
     excluded_fields = {
         "enable_trial_app",
         "is_allow_create_workspace",
@@ -372,7 +365,7 @@ def test_system_features_specs_exclude_backend_only_fields(tmp_path: Path):
     }
 
     for spec_name in ("console-openapi.json", "web-openapi.json"):
-        spec_path = next(path for path in written_paths if path.name == spec_name)
+        spec_path = next(path for path in generated_specs if path.name == spec_name)
         payload = json.loads(spec_path.read_text(encoding="utf-8"))
         schemas = payload["components"]["schemas"]
         system_features_schema = schemas["SystemFeatureModel"]
@@ -381,22 +374,15 @@ def test_system_features_specs_exclude_backend_only_fields(tmp_path: Path):
         assert "PluginManagerModel" not in schemas
 
 
-def test_generate_specs_writes_get_operations_without_request_bodies(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-
-    for path in written_paths:
+def test_generate_specs_writes_get_operations_without_request_bodies(generated_specs: list[Path]):
+    for path in generated_specs:
         payload = json.loads(path.read_text(encoding="utf-8"))
 
         assert all("requestBody" not in operation for operation in _get_operations(payload))
 
 
-def test_generate_specs_include_command_oriented_knowledge_fs_contract(tmp_path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-    openapi_path = next(path for path in written_paths if path.name == "openapi-openapi.json")
+def test_generate_specs_include_command_oriented_knowledge_fs_contract(generated_specs: list[Path]):
+    openapi_path = next(path for path in generated_specs if path.name == "openapi-openapi.json")
     payload = json.loads(openapi_path.read_text(encoding="utf-8"))
     paths = payload["paths"]
     prefix = "/workspaces/{workspace_id}/knowledge-fs/knowledge-spaces/{knowledge_space_id}/fs"
@@ -453,11 +439,8 @@ def test_generate_specs_include_command_oriented_knowledge_fs_contract(tmp_path)
     assert tree_operation["operationId"] == "tree_knowledge_fs"
 
 
-def test_generate_specs_writes_service_api_reference_descriptions(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-    service_path = next(path for path in written_paths if path.name == "service-openapi.json")
+def test_generate_specs_writes_service_api_reference_descriptions(generated_specs: list[Path]):
+    service_path = next(path for path in generated_specs if path.name == "service-openapi.json")
     payload = json.loads(service_path.read_text(encoding="utf-8"))
 
     chat_operation = payload["paths"]["/chat-messages"]["post"]
@@ -764,11 +747,8 @@ def test_generate_specs_writes_service_api_reference_descriptions(tmp_path: Path
     }
 
 
-def test_generate_specs_writes_web_file_upload_error_codes(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-    web_path = next(path for path in written_paths if path.name == "web-openapi.json")
+def test_generate_specs_writes_web_file_upload_error_codes(generated_specs: list[Path]):
+    web_path = next(path for path in generated_specs if path.name == "web-openapi.json")
     payload = json.loads(web_path.read_text(encoding="utf-8"))
 
     upload_bad_request = payload["paths"]["/files/upload"]["post"]["responses"]["400"]["description"]
@@ -803,11 +783,8 @@ def test_generate_specs_is_idempotent(tmp_path: Path):
         assert first_path.read_text(encoding="utf-8") == second_path.read_text(encoding="utf-8")
 
 
-def test_generate_specs_include_agent_v2_knowledge_set_schema_and_query_enums(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-    console_path = next(path for path in written_paths if path.name == "console-openapi.json")
+def test_generate_specs_include_agent_v2_knowledge_set_schema_and_query_enums(generated_specs: list[Path]):
+    console_path = next(path for path in generated_specs if path.name == "console-openapi.json")
     payload = json.loads(console_path.read_text(encoding="utf-8"))
     schemas = payload["components"]["schemas"]
 
@@ -818,11 +795,8 @@ def test_generate_specs_include_agent_v2_knowledge_set_schema_and_query_enums(tm
     assert schemas["AgentKnowledgeQueryMode"]["enum"] == ["generated_query", "user_query"]
 
 
-def test_generate_specs_include_console_contract_shapes_for_schema_migration(tmp_path: Path):
-    module = _load_generate_swagger_specs_module()
-
-    written_paths = module.generate_specs(tmp_path)
-    console_path = next(path for path in written_paths if path.name == "console-openapi.json")
+def test_generate_specs_include_console_contract_shapes_for_schema_migration(generated_specs: list[Path]):
+    console_path = next(path for path in generated_specs if path.name == "console-openapi.json")
     payload = json.loads(console_path.read_text(encoding="utf-8"))
     schemas = payload["components"]["schemas"]
     paths = payload["paths"]
diff --git a/api/tests/unit_tests/models/test_enums.py b/api/tests/unit_tests/models/test_enums.py
index 7db7debe122d2b..9884105cda4fa1 100644
--- a/api/tests/unit_tests/models/test_enums.py
+++ b/api/tests/unit_tests/models/test_enums.py
@@ -1,5 +1,7 @@
 import ast
 import inspect
+import os
+from functools import cache
 from pathlib import Path
 
 import pytest
@@ -12,6 +14,25 @@
 API_ROOT = Path(__file__).resolve().parents[3]
 
 
+@cache
+def _production_calls() -> tuple[tuple[Path, ast.Call], ...]:
+    """Every call expression in the API's production sources, with its path relative to ``API_ROOT``.
+
+    Parsed once for all source-scanning tests; ``tests`` and ``.venv`` directories are pruned
+    during the walk instead of being traversed and filtered afterwards.
+    """
+    calls: list[tuple[Path, ast.Call]] = []
+    for directory, subdirectories, filenames in os.walk(API_ROOT):
+        subdirectories[:] = [name for name in subdirectories if name not in {"tests", ".venv"}]
+        for filename in filenames:
+            if filename.endswith(".py"):
+                source_path = Path(directory) / filename
+                tree = ast.parse(source_path.read_text(), filename=str(source_path))
+                relative_path = source_path.relative_to(API_ROOT)
+                calls.extend((relative_path, node) for node in ast.walk(tree) if isinstance(node, ast.Call))
+    return tuple(calls)
+
+
 def test_end_user_type_covers_persisted_creation_values() -> None:
     assert {member.value for member in EndUserType} == {
         "app-deploy",
@@ -50,28 +71,21 @@ def test_end_user_service_callers_pass_end_user_type() -> None:
     method_names = {"get_or_create_end_user_by_type", "create_end_user_batch"}
     checked_calls = 0
 
-    for source_path in API_ROOT.rglob("*.py"):
-        if "tests" in source_path.parts or ".venv" in source_path.parts:
+    for relative_path, node in _production_calls():
+        if not isinstance(node.func, ast.Attribute) or node.func.attr not in method_names:
             continue
+        checked_calls += 1
 
-        tree = ast.parse(source_path.read_text(), filename=str(source_path))
-        for node in ast.walk(tree):
-            if not isinstance(node, ast.Call):
-                continue
-            if not isinstance(node.func, ast.Attribute) or node.func.attr not in method_names:
-                continue
-            checked_calls += 1
-
-            type_arg = next((keyword.value for keyword in node.keywords if keyword.arg == "type"), None)
-            if type_arg is None and node.args:
-                type_arg = node.args[0]
+        type_arg = next((keyword.value for keyword in node.keywords if keyword.arg == "type"), None)
+        if type_arg is None and node.args:
+            type_arg = node.args[0]
 
-            if not (
-                isinstance(type_arg, ast.Attribute)
-                and isinstance(type_arg.value, ast.Name)
-                and type_arg.value.id == "EndUserType"
-            ):
-                violations.append(f"{source_path.relative_to(API_ROOT)}:{node.lineno}")
+        if not (
+            isinstance(type_arg, ast.Attribute)
+            and isinstance(type_arg.value, ast.Name)
+            and type_arg.value.id == "EndUserType"
+        ):
+            violations.append(f"{relative_path}:{node.lineno}")
 
     assert checked_calls > 0
     assert violations == []
@@ -87,31 +101,24 @@ def test_end_user_type_column_uses_enum_text() -> None:
 def test_production_end_user_constructors_use_end_user_type_enum() -> None:
     violations: list[str] = []
 
-    for source_path in API_ROOT.rglob("*.py"):
-        if "tests" in source_path.parts or ".venv" in source_path.parts:
+    for relative_path, node in _production_calls():
+        if not isinstance(node.func, ast.Name) or node.func.id != "EndUser":
             continue
 
-        tree = ast.parse(source_path.read_text(), filename=str(source_path))
-        for node in ast.walk(tree):
-            if not isinstance(node, ast.Call):
+        for keyword in node.keywords:
+            if keyword.arg != "type":
                 continue
-            if not isinstance(node.func, ast.Name) or node.func.id != "EndUser":
-                continue
-
-            for keyword in node.keywords:
-                if keyword.arg != "type":
-                    continue
-                value = keyword.value
-                uses_end_user_type_member = (
-                    isinstance(value, ast.Attribute)
-                    and isinstance(value.value, ast.Name)
-                    and value.value.id == "EndUserType"
-                )
-                uses_end_user_type_conversion = (
-                    isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id == "EndUserType"
-                )
-                if not (uses_end_user_type_member or uses_end_user_type_conversion):
-                    violations.append(f"{source_path.relative_to(API_ROOT)}:{node.lineno}")
+            value = keyword.value
+            uses_end_user_type_member = (
+                isinstance(value, ast.Attribute)
+                and isinstance(value.value, ast.Name)
+                and value.value.id == "EndUserType"
+            )
+            uses_end_user_type_conversion = (
+                isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id == "EndUserType"
+            )
+            if not (uses_end_user_type_member or uses_end_user_type_conversion):
+                violations.append(f"{relative_path}:{node.lineno}")
 
     assert violations == []
 
diff --git a/api/tests/unit_tests/services/test_conversation_query_service.py b/api/tests/unit_tests/services/test_conversation_query_service.py
index 758eb5d7e731db..bf57c3454391b4 100644
--- a/api/tests/unit_tests/services/test_conversation_query_service.py
+++ b/api/tests/unit_tests/services/test_conversation_query_service.py
@@ -616,7 +616,7 @@ def do_HEAD(self) -> None:
                 self.end_headers()
 
         server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-        thread = Thread(target=server.serve_forever, daemon=True)
+        thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
         thread.start()
         try:
             with sqlite_session_factory.begin() as session:
diff --git a/api/tests/unit_tests/services/test_message_query_adapters.py b/api/tests/unit_tests/services/test_message_query_adapters.py
index f32b184b670908..ec1b86422fb96f 100644
--- a/api/tests/unit_tests/services/test_message_query_adapters.py
+++ b/api/tests/unit_tests/services/test_message_query_adapters.py
@@ -171,7 +171,7 @@ def do_HEAD(self) -> None:
     http_pool = http_client_pooling.HttpClientPoolFactory()
     monkeypatch.setattr(http_client_pooling, "_factory", http_pool)
     server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-    thread = Thread(target=server.serve_forever, daemon=True)
+    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     thread.start()
     event.listen(sqlite_session_factory, "after_begin", observe)
     event.listen(sqlite_session_factory, "after_commit", committed)