fix: preserve SDK behavior while closing runtime resources

.github/workflows/test.yml

@@ -31,11 +31,14 @@ jobs:
       - name: Install just
         uses: extractions/setup-just@53165ef7e734c5c07cb06b3c8e7b647c5aa16db3 # v4
 
-      - name: Setup dify-plugin-cli
-        run: ./scripts/setup-dify-plugin-cli.sh
-
-      - name: Install dependencies
-        run: just dev
-
-      - name: Run tests
-        run: just test
+      - name: Install dependencies and run tests
+        run: |
+          uv sync --locked
+          just coverage
+
+      - name: Upload test reports
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: test-reports-${{ matrix.python-version }}
+          path: reports/

.gitignore

@@ -9,6 +9,10 @@ dist/
 
 # Unit test
 .pytest_cache/
+/.hypothesis/
+/.coverage
+/.coverage.*
+/reports/
 
 # Environments
 .venv/

CONTRIBUTING.md

@@ -35,7 +35,8 @@ source .venv/bin/activate
 
 The repository uses [`uv`](https://docs.astral.sh/uv/) for dependency and
 virtual environment management. The default development environment includes
-`pytest`, `pytest-cov`, `pytest-mock`, `pytest-xprocess`, `ruff`, `ty`, and
+`pytest`, `pytest-cov`, `pytest-mock`, `pytest-httpserver`, `pytest-timeout`,
+`hypothesis`, `trustme`, `ruff`, `ty`, and
 [`prek`](https://prek.j178.dev/).
 
 ### Git Hooks
@@ -65,7 +66,7 @@ Use `just` by default. For targeted work, direct tool usage is still fine:
 
 ```bash
 uv run ruff check src/dify_plugin/path.py
-uv run pytest tests/path/test_file.py -k keyword
+just test tests/path/test_file.py -k keyword
 uv run prek run -a
 ```
 
@@ -77,15 +78,14 @@ Use these commands for normal development:
 - `just lint`: run `just fmt`, then `uv run ruff check --fix`
 - `just check`: run `uv lock --check`, `ruff format --check --diff`, and `ruff check`
 - `just test`: run the SDK tests
+- `just coverage`: run tests with subprocess and branch coverage, then enforce the coverage gates
 - `just build`: build source and wheel distributions
 - `just docs`: generate schema documentation into `.mkdocs/docs/schema.md`
 - `just clean`: remove local build, test, and lint artifacts
 
 Notes:
 
 - `just check` is the non-mutating validation entrypoint used by PR checks and Git hooks.
-- Integration tests that need the Dify plugin CLI are skipped when the binary is unavailable.
-- CI installs the CLI with [`scripts/setup-dify-plugin-cli.sh`](scripts/setup-dify-plugin-cli.sh) before running tests.
 - If you change dependencies, refresh and commit [`uv.lock`](uv.lock) before opening a pull request.
 
 Run `just check` for every change, plus `just test` for behavior changes, `just build` for packaging changes, and `just docs` for schema changes.
@@ -96,9 +96,7 @@ Pull requests targeting `main` currently run these checks:
 
 1. PR title validation with `amannn/action-semantic-pull-request`
 2. `just check` on Python 3.12, including `uv.lock` freshness validation
-3. `just test` on Python 3.12 and 3.13 through the reusable test workflow
-
-The test workflow installs the Dify plugin CLI, runs `just dev`, and then runs `just test`.
+3. `just coverage` on Python 3.12 and 3.13 through the reusable test workflow
 
 Pushes to `main` also run the MkDocs workflow.
 It runs `just docs` on Python 3.12 and deploys `.mkdocs` to GitHub Pages.

justfile

@@ -15,8 +15,16 @@ check:
     uv run ruff format --check --diff
     uv run ruff check
 
-test:
-    uv run pytest
+[positional-arguments]
+test *args:
+    uv run python -m tests "$@"
+
+coverage:
+    uv run python -m tests --cov \
+        --cov-report=json:reports/coverage.json \
+        --cov-report=xml:reports/coverage.xml \
+        --junitxml=reports/junit.xml
+    uv run python scripts/check_coverage.py reports/coverage.json
 
 build:
     uv build --no-create-gitignore --no-sources

pyproject.toml

@@ -35,12 +35,15 @@ build-backend = 'uv_build'
 
 [dependency-groups]
 dev = [
+    "hypothesis>=6.169.0",
     'prek',
     'pytest',
     'pytest-cov',
+    "pytest-httpserver>=1.2.0",
     'pytest-mock',
-    'pytest-xprocess',
+    "pytest-timeout>=2.4.0",
     'ruff',
+    "trustme>=1.2.1",
     'ty',
 ]
 
@@ -50,6 +53,27 @@ default-groups = ['dev']
 [tool.pytest]
 minversion = '9.0'
 testpaths = ['tests']
+addopts = ['--strict-config', '--strict-markers']
+# pytest-timeout registers a string ini value, even with pytest's native TOML config.
+timeout = '60'
+xfail_strict = true
+markers = ['integration: runs a real plugin process or local protocol service']
+filterwarnings = [
+    'error::ResourceWarning',
+    'error::pytest.PytestUnhandledThreadExceptionWarning',
+    'error::pytest.PytestUnraisableExceptionWarning',
+]
+
+[tool.coverage.run]
+branch = true
+source_pkgs = ['dify_plugin']
+# SDK threads are monkey-patched greenlets; a second thread tracer drops events.
+concurrency = ['gevent']
+patch = ['subprocess']
+sigterm = true
+
+[tool.coverage.paths]
+source = ['src/dify_plugin', '*/site-packages/dify_plugin']
 
 [tool.ruff]
 preview = true

scripts/check_coverage.py

@@ -0,0 +1,58 @@
+import json
+import sys
+from pathlib import Path
+
+GATES = (
+    ("SDK", ("src/dify_plugin/",), 85, 75),
+    (
+        "runtime",
+        ("src/dify_plugin/core/runtime.py", "src/dify_plugin/core/server/"),
+        90,
+        85,
+    ),
+    (
+        "HTTP",
+        (
+            "src/dify_plugin/interfaces/model/openai_compatible/",
+            "src/dify_plugin/file/",
+            "src/dify_plugin/invocations/file.py",
+        ),
+        90,
+        85,
+    ),
+)
+
+
+def check_coverage(report: dict) -> list[str]:
+    if not report["meta"]["branch_coverage"]:
+        return ["Branch coverage was not collected"]
+    failures = []
+    for name, prefixes, minimum_lines, minimum_branches in GATES:
+        summaries = [
+            data["summary"]
+            for path, data in report["files"].items()
+            if path.startswith(prefixes)
+        ]
+        if not summaries:
+            failures.append(f"{name}: no matching source files in coverage report")
+            continue
+        for metric, covered_key, total_key, minimum in (
+            ("lines", "covered_lines", "num_statements", minimum_lines),
+            ("branches", "covered_branches", "num_branches", minimum_branches),
+        ):
+            covered = sum(summary[covered_key] for summary in summaries)
+            total = sum(summary[total_key] for summary in summaries)
+            percent = 100 * covered / total if total else 0
+            sys.stdout.write(f"{name} {metric}: {percent:.2f}% (minimum {minimum}%)\n")
+            if percent < minimum:
+                failures.append(
+                    f"{name} {metric}: {covered}/{total} ({percent:.4f}%) < {minimum}%"
+                )
+    return failures
+
+
+if __name__ == "__main__":
+    failures = check_coverage(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")))
+    if failures:
+        sys.stderr.write("\n".join(failures) + "\n")
+    raise SystemExit(bool(failures))

src/dify_plugin/core/documentation/generator.py

@@ -38,7 +38,6 @@ def __init__(self) -> None:
         self._processed_types: set[type] = set()
         self._field_descriptions: dict[tuple[type, str], str] = {}
         self._schema_descriptions: dict[type, str] = {}
-        self._processed_field_types: set[type] = set()
         self._type_to_schema: dict[type, object] = {}
         self._type_blocks: dict[type, int] = {}
         self._blocks: list[list] = []
@@ -186,15 +185,15 @@ def _preprocess_schemas(self, schemas: list) -> None:
         # First pass: collect all field descriptions
         for schema in schemas:
             cls = schema.cls
-            if not issubclass(cls, BaseModel):
-                continue
-
             # Store schema description
             if cls not in self._schema_descriptions or len(schema.description) > len(
                 self._schema_descriptions[cls],
             ):
                 self._schema_descriptions[cls] = schema.description
 
+            if not issubclass(cls, BaseModel):
+                continue
+
             # Store field descriptions
             outside_reference_fields = (
                 getattr(schema, "outside_reference_fields", {}) or {}
@@ -381,12 +380,6 @@ def _write_schema_doc(self, f: TextIO, type_: type) -> None:
                 if field_type is None:
                     continue
 
-                # Skip if we've already processed this field type
-                if isinstance(field_type, type) and issubclass(field_type, BaseModel):
-                    if field_type in self._processed_field_types:
-                        continue
-                    self._processed_field_types.add(field_type)
-
                 # Skip if we've already processed this field
                 field_key = (field_type, field_name)
                 if field_key in processed_fields:

src/dify_plugin/core/plugin_executor.py

@@ -694,42 +694,45 @@ def invoke_endpoint(
         except Exception as e:
             response = Response(f"Internal Server Error: {e!s}", status=500)
 
-        # check if response is a generator
-        if isinstance(response.response, Generator):
-            # return headers
-            yield {
-                "status": response.status_code,
-                "headers": dict(response.headers.items()),
-            }
-
-            for chunk in response.response:
-                if isinstance(chunk, bytes | bytearray | memoryview):
-                    yield {"result": binascii.hexlify(chunk).decode()}
-                else:
-                    yield {"result": binascii.hexlify(chunk.encode("utf-8")).decode()}
-        else:
-            result = {
-                "status": response.status_code,
-                "headers": dict(response.headers.items()),
-            }
+        with response:
+            # check if response is a generator
+            if isinstance(response.response, Generator):
+                # return headers
+                yield {
+                    "status": response.status_code,
+                    "headers": dict(response.headers.items()),
+                }
 
-            if isinstance(response.response, bytes | bytearray | memoryview):
-                result["result"] = binascii.hexlify(response.response).decode()
-            elif isinstance(response.response, str):
-                result["result"] = binascii.hexlify(
-                    response.response.encode("utf-8"),
-                ).decode()
-            elif isinstance(response.response, Iterable):
-                result["result"] = ""
                 for chunk in response.response:
                     if isinstance(chunk, bytes | bytearray | memoryview):
-                        result["result"] += binascii.hexlify(chunk).decode()
+                        yield {"result": binascii.hexlify(chunk).decode()}
                     else:
-                        result["result"] += binascii.hexlify(
-                            chunk.encode("utf-8"),
-                        ).decode()
-
-            yield result
+                        yield {
+                            "result": binascii.hexlify(chunk.encode("utf-8")).decode()
+                        }
+            else:
+                result = {
+                    "status": response.status_code,
+                    "headers": dict(response.headers.items()),
+                }
+
+                if isinstance(response.response, bytes | bytearray | memoryview):
+                    result["result"] = binascii.hexlify(response.response).decode()
+                elif isinstance(response.response, str):
+                    result["result"] = binascii.hexlify(
+                        response.response.encode("utf-8"),
+                    ).decode()
+                elif isinstance(response.response, Iterable):
+                    result["result"] = ""
+                    for chunk in response.response:
+                        if isinstance(chunk, bytes | bytearray | memoryview):
+                            result["result"] += binascii.hexlify(chunk).decode()
+                        else:
+                            result["result"] += binascii.hexlify(
+                                chunk.encode("utf-8"),
+                            ).decode()
+
+                yield result
 
     def _get_oauth_provider_instance(
         self,
@@ -966,14 +969,15 @@ def dispatch_trigger_event(
             subscription=subscription,
             request=original_request,
         )
-        return TriggerDispatchResponse(
-            user_id=dispatch_result.user_id,
-            events=dispatch_result.events,
-            response=binascii.hexlify(
-                data=serialize_response(response=dispatch_result.response),
-            ).decode(),
-            payload=dispatch_result.payload,
-        )
+        with dispatch_result.response:
+            return TriggerDispatchResponse(
+                user_id=dispatch_result.user_id,
+                events=dispatch_result.events,
+                response=binascii.hexlify(
+                    data=serialize_response(response=dispatch_result.response),
+                ).decode(),
+                payload=dispatch_result.payload,
+            )
 
     def subscribe_trigger(
         self,

src/dify_plugin/core/plugin_registration.py

@@ -324,16 +324,32 @@ def _resolve_trigger_providers(self) -> None:
                 parent_type=Trigger,
             )
 
+            constructor_source = source
             subscription_constructor_cls_candidates = load_multi_subclasses_from_source(
                 module_name=module_source,
                 script_path=_source_to_script_path(source),
                 parent_type=TriggerSubscriptionConstructor,
             )
+            constructor = provider.subscription_constructor
+            if (
+                not subscription_constructor_cls_candidates
+                and constructor
+                and constructor.extra
+                and constructor.extra.python.source != source
+            ):
+                constructor_source = constructor.extra.python.source
+                subscription_constructor_cls_candidates = (
+                    load_multi_subclasses_from_source(
+                        module_name=_source_to_module_name(constructor_source),
+                        script_path=_source_to_script_path(constructor_source),
+                        parent_type=TriggerSubscriptionConstructor,
+                    )
+                )
 
             if len(subscription_constructor_cls_candidates) > 1:
                 msg = (
                     "Multiple TriggerSubscriptionConstructor subclasses found "
-                    f"in {source}."
+                    f"in {constructor_source}."
                     " Only a single implementation is supported."
                 )
                 raise ValueError(msg)
@@ -350,7 +366,7 @@ def _resolve_trigger_providers(self) -> None:
             ):
                 msg = (
                     "Trigger subscription constructor configuration declared "
-                    f"but no implementation found in {source}."
+                    f"but no implementation found in {constructor_source}."
                 )
                 raise ValueError(msg)