From ee88113c620acec3ef1cdfa6cdfcb131a9be2cb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=97=E7=8E=AE=20=28Jade=20Lin=29?= Date: Wed, 11 Nov 2026 14:01:34 +0800 Subject: [PATCH] fix(api): use direct dispatch for hosted moderation fallback --- api/core/plugin/impl/model.py | 49 ++++++++ api/services/moderation_adapters.py | 30 ++++- api/services/plugin/dispatch_identity.py | 12 +- .../unit/core/model_moderation_fixtures.py | 33 +++++- .../unit_tests/core/plugin/impl/test_model.py | 98 +++++++++++++++ .../services/plugin/test_dispatch_identity.py | 17 ++- .../services/test_moderation_adapters.py | 112 ++++++++++++++++++ docs/BACKEND_MODERATION.md | 21 +++- 8 files changed, 353 insertions(+), 19 deletions(-) diff --git a/api/core/plugin/impl/model.py b/api/core/plugin/impl/model.py index 477260608a6..49475b1a538 100644 --- a/api/core/plugin/impl/model.py +++ b/api/core/plugin/impl/model.py @@ -730,3 +730,52 @@ class PluginModelClient(BasePluginClient): return resp.result raise ValueError("Failed to invoke moderation") + + def invoke_moderation_direct( + self, + tenant_id: str, + user_id: str | None, + plugin_id: str, + provider: str, + model: str, + credentials: dict[str, Any], + text: str, + *, + plugin_unique_identifier: str, + ) -> bool: + """Invoke a server-selected package without requiring a tenant installation. + + Preconditions: The caller admits the tenant and selects a trusted pinned + package whose runtime is already available. The daemon server key remains + mandatory. Postconditions: The JSON payload retains tenant/user context; + the identifier is mandatory and must match plugin_id. Daemon and empty + response failures propagate without installation or credential fallback. + """ + if not tenant_id.strip(): + raise ValueError("A tenant is required for direct moderation dispatch") + if not plugin_unique_identifier.strip(): + raise ValueError("A plugin unique identifier is required for direct moderation dispatch") + response = self._request_with_plugin_daemon_response_stream( + method="POST", + path="v2/invoke/dispatch/moderation/invoke", + type_=PluginBasicBooleanResponse, + data=jsonable_encoder( + { + "tenant_id": tenant_id, + **self._dispatch_payload( + user_id=user_id, + data={ + "provider": provider, + "model_type": "moderation", + "model": model, + "credentials": credentials, + "text": text, + }, + ), + } + ), + headers=plugin_dispatch_headers(plugin_id, plugin_unique_identifier), + ) + for result in response: + return result.result + raise ValueError("Failed to invoke moderation") diff --git a/api/services/moderation_adapters.py b/api/services/moderation_adapters.py index 5689ac002ea..b3bd721bfe5 100644 --- a/api/services/moderation_adapters.py +++ b/api/services/moderation_adapters.py @@ -47,7 +47,7 @@ logger = logging.getLogger(__name__) class HostedModerationPluginClient(Protocol): - """Direct daemon dispatch surface for a pinned hosted moderation plugin.""" + """Keep installed-tenant dispatch distinct from server-selected direct fallback.""" def invoke_moderation( self, @@ -62,6 +62,19 @@ class HostedModerationPluginClient(Protocol): plugin_unique_identifier: str | None = None, ) -> bool: ... + def invoke_moderation_direct( + self, + tenant_id: str, + user_id: str | None, + plugin_id: str, + provider: str, + model: str, + credentials: dict[str, Any], + text: str, + *, + plugin_unique_identifier: str, + ) -> bool: ... + class TenantModerationPluginResolver(Protocol): """Resolve an installed official, verified OpenAI plugin in the supplied tenant.""" @@ -134,14 +147,19 @@ class HostedOpenAIModerationGateway: if not valid_package or plugin_id != "langgenius/openai": raise ValueError("Hosted moderation requires a pinned langgenius/openai plugin identifier.") - def _invoke(self, context: AppRequestContext, text: str, identifier: str) -> bool: - """Invoke one admitted package with platform credentials; reject malformed verdicts.""" + def _invoke(self, context: AppRequestContext, text: str, identifier: str, *, direct: bool) -> bool: + """Invoke an admitted package; direct fallback never resolves a tenant installation. + + The caller selects direct dispatch only for the SRE-pinned fallback. + Both routes retain platform credentials, tenant context, and strict verdict validation. + """ self._validate_identifier(identifier) if self._credentials is None: raise InvokeAuthorizationError("Hosted moderation credentials are not configured.") if not text: return False - flagged = self._client.invoke_moderation( + invoke = self._client.invoke_moderation_direct if direct else self._client.invoke_moderation + flagged = invoke( tenant_id=context.tenant_id, user_id=None, plugin_id="langgenius/openai", @@ -163,7 +181,7 @@ class HostedOpenAIModerationGateway: self._tenant_identifiers[context.tenant_id] = self._tenant_plugins.resolve(context.tenant_id) identifier = self._tenant_identifiers[context.tenant_id] if identifier is not None: - return self._invoke(context, text, identifier), "tenant" + return self._invoke(context, text, identifier, direct=False), "tenant" except Exception as error: self._tenant_identifiers[context.tenant_id] = None logger.warning( @@ -176,7 +194,7 @@ class HostedOpenAIModerationGateway: raise InvokeAuthorizationError( "No trusted tenant OpenAI plugin or configured moderation fallback is available." ) - return self._invoke(context, text, self._plugin_unique_identifier), "configured" + return self._invoke(context, text, self._plugin_unique_identifier, direct=True), "configured" def check( self, context: AppRequestContext, *, rule: ModerationRule, request: ModerationRequest, phase: ModerationPhase diff --git a/api/services/plugin/dispatch_identity.py b/api/services/plugin/dispatch_identity.py index 071b758ed24..78824b80035 100644 --- a/api/services/plugin/dispatch_identity.py +++ b/api/services/plugin/dispatch_identity.py @@ -29,12 +29,22 @@ def plugin_dispatch_headers(plugin_id: str, identifier: str | None) -> dict[str, def prepare_dispatch_identity(path: str, headers: dict[str, str], identifier: str | None) -> None: + """Bind matching identities for tenant and direct dispatch without installation lookup. + + Legacy tenant routes may omit an identifier. Direct routes require an exact + identifier and plugin ID before any request can send server-owned credentials. + Non-dispatch requests remain unchanged. + """ parts = path.split("/") - if len(parts) < 4 or parts[0] != "plugin" or parts[2] != "dispatch": + tenant_dispatch = len(parts) >= 4 and parts[0] == "plugin" and parts[2] == "dispatch" + direct_dispatch = parts[:3] == ["v2", "invoke", "dispatch"] + if not tenant_dispatch and not direct_dispatch: return identifier = headers.get(PLUGIN_IDENTIFIER_HEADER) or identifier if identifier is None: + if direct_dispatch: + raise ValueError("A plugin unique identifier is required for direct dispatch") # Legacy calls leave installation resolution to the daemon. return diff --git a/api/tests/helpers/unit/core/model_moderation_fixtures.py b/api/tests/helpers/unit/core/model_moderation_fixtures.py index 3332c6236ea..66bd5daed5a 100644 --- a/api/tests/helpers/unit/core/model_moderation_fixtures.py +++ b/api/tests/helpers/unit/core/model_moderation_fixtures.py @@ -248,10 +248,11 @@ class RecordingPollingRuntime(PluginModelRuntime): class RecordingHostedModerationClient(PluginModelClient): - """Record direct identifier dispatch without tenant provider discovery or daemon I/O.""" + """Record installed and direct dispatch without tenant provider discovery or daemon I/O.""" def __init__(self) -> None: self.calls: list[tuple[str, str, str, str, dict[str, Any], str, str | None]] = [] + self.routes: list[str] = [] self.error: Exception | None = None self.identifier_errors: dict[str, Exception] = {} @@ -266,6 +267,36 @@ class RecordingHostedModerationClient(PluginModelClient): credentials: dict[str, Any], text: str, plugin_unique_identifier: str | None = None, + ) -> bool: + self.routes.append("tenant") + return self._record(tenant_id, user_id, plugin_id, provider, model, credentials, text, plugin_unique_identifier) + + @override + def invoke_moderation_direct( + self, + tenant_id: str, + user_id: str | None, + plugin_id: str, + provider: str, + model: str, + credentials: dict[str, Any], + text: str, + *, + plugin_unique_identifier: str, + ) -> bool: + self.routes.append("direct") + return self._record(tenant_id, user_id, plugin_id, provider, model, credentials, text, plugin_unique_identifier) + + def _record( + self, + tenant_id: str, + user_id: str | None, + plugin_id: str, + provider: str, + model: str, + credentials: dict[str, Any], + text: str, + plugin_unique_identifier: str | None, ) -> bool: self.calls.append( ( diff --git a/api/tests/unit_tests/core/plugin/impl/test_model.py b/api/tests/unit_tests/core/plugin/impl/test_model.py index 36e695096c6..df03d038bde 100644 --- a/api/tests/unit_tests/core/plugin/impl/test_model.py +++ b/api/tests/unit_tests/core/plugin/impl/test_model.py @@ -35,6 +35,104 @@ from graphon.model_runtime.entities.provider_entities import ConfigurateMethod, PLUGIN_IDENTIFIER = "org/plugin:1.0.0@aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +@pytest.mark.parametrize("user_id", [None, "user-1"]) +@pytest.mark.parametrize("flagged", [False, True]) +def test_direct_moderation_sends_explicit_tenant_and_user_context( + monkeypatch: pytest.MonkeyPatch, user_id: str | None, flagged: bool +) -> None: + requests: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, text="data: " + json.dumps({"code": 0, "message": "", "data": {"result": flagged}})) + + with httpx.Client(transport=httpx.MockTransport(handle)) as http_client: + monkeypatch.setattr(client_module, "_httpx_client", http_client) + assert ( + PluginModelClient().invoke_moderation_direct( + tenant_id="tenant-1", + user_id=user_id, + plugin_id="org/plugin", + provider="provider-a", + model="moderation-a", + credentials={"api_key": "platform-key"}, + text="content", + plugin_unique_identifier=PLUGIN_IDENTIFIER, + ) + is flagged + ) + + assert len(requests) == 1 + assert requests[0].url.path == "/v2/invoke/dispatch/moderation/invoke" + assert requests[0].headers["X-Plugin-ID"] == "org/plugin" + assert requests[0].headers["X-Plugin-Unique-Identifier"] == PLUGIN_IDENTIFIER + assert "X-Api-Key" in requests[0].headers + payload = json.loads(requests[0].content) + assert payload["tenant_id"] == "tenant-1" + assert payload.get("user_id") == user_id + assert payload["data"] == { + "provider": "provider-a", + "model_type": "moderation", + "model": "moderation-a", + "credentials": {"api_key": "platform-key"}, + "text": "content", + } + + +@pytest.mark.parametrize( + ("tenant_id", "identifier"), + [ + ("", PLUGIN_IDENTIFIER), + (" \t", PLUGIN_IDENTIFIER), + ("tenant-1", ""), + ("tenant-1", " \t"), + ("tenant-1", "org/other:1.0.0@checksum"), + ], +) +def test_direct_moderation_rejects_missing_scope_or_mismatched_identity_before_http( + monkeypatch: pytest.MonkeyPatch, tenant_id: str, identifier: str +) -> None: + def unexpected_request(_request: httpx.Request) -> httpx.Response: + raise AssertionError("Invalid direct dispatch must not send credentials") + + with httpx.Client(transport=httpx.MockTransport(unexpected_request)) as http_client: + monkeypatch.setattr(client_module, "_httpx_client", http_client) + with pytest.raises(ValueError, match="required|does not match"): + PluginModelClient().invoke_moderation_direct( + tenant_id=tenant_id, + user_id=None, + plugin_id="org/plugin", + provider="provider-a", + model="moderation-a", + credentials={"api_key": "platform-key"}, + text="content", + plugin_unique_identifier=identifier, + ) + + +def test_direct_moderation_rejects_an_empty_daemon_response(monkeypatch: pytest.MonkeyPatch) -> None: + requests: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, text="") + + with httpx.Client(transport=httpx.MockTransport(handle)) as http_client: + monkeypatch.setattr(client_module, "_httpx_client", http_client) + with pytest.raises(ValueError, match="Failed to invoke moderation"): + PluginModelClient().invoke_moderation_direct( + tenant_id="tenant-1", + user_id=None, + plugin_id="org/plugin", + provider="provider-a", + model="moderation-a", + credentials={}, + text="content", + plugin_unique_identifier=PLUGIN_IDENTIFIER, + ) + assert len(requests) == 1 + + class TestModelClientCases: class TestPluginModelClient: def test_fetch_model_providers(self, mocker: MockerFixture): diff --git a/api/tests/unit_tests/services/plugin/test_dispatch_identity.py b/api/tests/unit_tests/services/plugin/test_dispatch_identity.py index 1065e29e75b..3eeb9b283b7 100644 --- a/api/tests/unit_tests/services/plugin/test_dispatch_identity.py +++ b/api/tests/unit_tests/services/plugin/test_dispatch_identity.py @@ -57,16 +57,21 @@ def test_unbound_dispatch_keeps_legacy_headers_without_refetch(monkeypatch: pyte assert headers == {"X-Plugin-ID": PLUGIN_ID} -def test_dispatch_rejects_cross_plugin_identity() -> None: +@pytest.mark.parametrize("path", ["plugin/tenant-a/dispatch/tool/invoke", "v2/invoke/dispatch/moderation/invoke"]) +def test_dispatch_rejects_cross_plugin_identity(path: str) -> None: with pytest.raises(ValueError, match="does not match"): - prepare_dispatch_identity( - "plugin/tenant-a/dispatch/tool/invoke", {"X-Plugin-ID": "langgenius/other"}, IDENTIFIER - ) + prepare_dispatch_identity(path, {"X-Plugin-ID": "langgenius/other"}, IDENTIFIER) -def test_pinned_dispatch_requires_plugin_id() -> None: +@pytest.mark.parametrize("path", ["plugin/tenant-a/dispatch/tool/invoke", "v2/invoke/dispatch/moderation/invoke"]) +def test_pinned_dispatch_requires_plugin_id(path: str) -> None: with pytest.raises(ValueError, match="X-Plugin-ID is required"): - prepare_dispatch_identity("plugin/tenant-a/dispatch/tool/invoke", {}, IDENTIFIER) + prepare_dispatch_identity(path, {}, IDENTIFIER) + + +def test_direct_dispatch_cannot_fall_back_to_installation_resolution() -> None: + with pytest.raises(ValueError, match="identifier is required"): + prepare_dispatch_identity("v2/invoke/dispatch/moderation/invoke", {"X-Plugin-ID": PLUGIN_ID}, None) def test_explicit_empty_identifier_cannot_fall_back_to_legacy_dispatch() -> None: diff --git a/api/tests/unit_tests/services/test_moderation_adapters.py b/api/tests/unit_tests/services/test_moderation_adapters.py index d7f4424e810..a4fffbef9da 100644 --- a/api/tests/unit_tests/services/test_moderation_adapters.py +++ b/api/tests/unit_tests/services/test_moderation_adapters.py @@ -1,7 +1,10 @@ +import json from collections.abc import Callable, Generator, Iterator from threading import Thread +from typing import Literal from uuid import uuid4 +import httpx import pytest from flask import Flask, request from sqlalchemy import Engine @@ -12,6 +15,8 @@ from core.app.apps.exc import AppGenerateError from core.app.llm.quota import ModelQuotaReservationState from core.plugin.entities.plugin import PluginInstallationSource from core.plugin.entities.plugin_daemon import PluginModelProviderDeclaration +from core.plugin.impl import base as client_module +from core.plugin.impl.model import PluginModelClient from enums import DeploymentEdition from extensions.application_services.moderation import build_llm_moderation_guard, build_moderation_service from extensions.ext_code_based_extension import code_based_extension @@ -154,6 +159,7 @@ def test_platform_prefers_tenant_plugin_and_logs_rejection_without_content( assert len(client.calls) == 1 assert client.calls[0][6] == (identifier or fallback) assert client.calls[0][4] == {"api_key": "platform-key"} + assert client.routes == (["direct"] if identifier is None else ["tenant"]) assert ("Platform moderation rejected content" in caplog.text) is flagged assert text not in caplog.text assert "platform-key" not in caplog.text @@ -197,6 +203,7 @@ def test_tenant_invoke_failure_switches_to_configured_fallback_for_remaining_che assert not gateway.check(context, rule=rule, request=ModerationRequest(text="safe"), phase="input").flagged assert not gateway.check(context, rule=rule, request=ModerationRequest(text="safe again"), phase="input").flagged assert [call[6] for call in client.calls] == [tenant_identifier, fallback, fallback] + assert client.routes == ["tenant", "direct", "direct"] assert plugins.tenants == ["tenant-1"] @@ -217,6 +224,111 @@ def test_trusted_tenant_plugin_does_not_require_a_valid_fallback_identifier() -> phase="input", ).flagged assert client.calls[0][6] == tenant_identifier + assert client.routes == ["tenant"] + + +@pytest.mark.parametrize("tenant_route", ["missing", "available", "unavailable"]) +@pytest.mark.parametrize("flagged", [False, True]) +@pytest.mark.parametrize("tenant_id", ["tenant-1", "tenant-2"]) +def test_hosted_moderation_keeps_tenant_installation_dispatch_separate_from_direct_fallback( + monkeypatch: pytest.MonkeyPatch, + config_overrides: Callable[..., None], + tenant_route: Literal["missing", "available", "unavailable"], + flagged: bool, + tenant_id: str, +) -> None: + config_overrides(PLUGIN_DAEMON_KEY="daemon-server-key") + tenant_identifier = "langgenius/openai:1.0.0@" + "a" * 32 + fallback_identifier = "langgenius/openai:2.0.0@" + "b" * 32 + requests: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + assert request.method == "POST" + assert request.headers["X-Api-Key"] == "daemon-server-key" + assert request.headers["X-Plugin-ID"] == "langgenius/openai" + payload = json.loads(request.content) + assert payload["data"] == { + "provider": "openai", + "model_type": "moderation", + "model": "omni-moderation-latest", + "credentials": {"openai_api_key": "platform-key"}, + "text": "content", + } + if request.url.path.startswith("/plugin/"): + assert request.url.path == f"/plugin/{tenant_id}/dispatch/moderation/invoke" + assert request.headers["X-Plugin-Unique-Identifier"] == tenant_identifier + if tenant_route != "available": + return httpx.Response(404, json={"code": -404, "message": "Plugin not installed", "data": None}) + else: + assert request.url.path == "/v2/invoke/dispatch/moderation/invoke" + assert request.headers["X-Plugin-Unique-Identifier"] == fallback_identifier + assert payload["tenant_id"] == tenant_id + assert "user_id" not in payload + return httpx.Response(200, text="data: " + json.dumps({"code": 0, "message": "", "data": {"result": flagged}})) + + with httpx.Client(transport=httpx.MockTransport(handle)) as http_client: + monkeypatch.setattr(client_module, "_httpx_client", http_client) + gateway = HostedOpenAIModerationGateway( + client=PluginModelClient(), + credentials={"openai_api_key": "platform-key"}, + plugin_unique_identifier=fallback_identifier, + tenant_plugins=RecordingTenantPlugins(None if tenant_route == "missing" else tenant_identifier), + ) + verdict = gateway.check( + AppRequestContext(tenant_id=tenant_id, app_id="app-1"), + rule=ModerationRule(type="openai_moderation", config={"inputs_config": {"enabled": True}}), + request=ModerationRequest(text="content"), + phase="input", + ) + assert verdict.flagged is flagged + paths = [request.url.path for request in requests] + tenant_path = f"/plugin/{tenant_id}/dispatch/moderation/invoke" + direct_path = "/v2/invoke/dispatch/moderation/invoke" + assert ( + paths + == { + "missing": [direct_path], + "available": [tenant_path], + "unavailable": [tenant_path, direct_path], + }[tenant_route] + ) + + +@pytest.mark.parametrize("status_code", [404, 500]) +def test_unavailable_direct_fallback_fails_closed_without_trying_a_tenant_provider( + monkeypatch: pytest.MonkeyPatch, status_code: int +) -> None: + requests: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + assert request.url.path == "/v2/invoke/dispatch/moderation/invoke" + return httpx.Response(status_code, json={"code": -1, "message": "Runtime unavailable", "data": None}) + + with httpx.Client(transport=httpx.MockTransport(handle)) as http_client: + monkeypatch.setattr(client_module, "_httpx_client", http_client) + guard = build_llm_moderation_guard( + AppRequestContext(tenant_id="tenant-1", app_id="app-1"), + service=ModerationService( + gateway=HostedOpenAIModerationGateway( + client=PluginModelClient(), + credentials={"openai_api_key": "platform-key"}, + plugin_unique_identifier="langgenius/openai:1.0.0@" + "a" * 32, + tenant_plugins=RecordingTenantPlugins(None), + ) + ), + rule=ModerationRule( + type="openai_moderation", + config={"inputs_config": {"enabled": True}, "outputs_config": {"enabled": True}}, + ), + platform=True, + ) + with pytest.raises(ModerationUnavailableInvocationError) as unavailable: + guard.check_inputs([UserPromptMessage(content="content")]) + assert unavailable.value.status_code == 503 + assert str(unavailable.value) == "This message can’t be processed right now. Please try again later." + assert len(requests) == 1 @pytest.fixture diff --git a/docs/BACKEND_MODERATION.md b/docs/BACKEND_MODERATION.md index 5ae178dbd2e..12a478fa4cf 100644 --- a/docs/BACKEND_MODERATION.md +++ b/docs/BACKEND_MODERATION.md @@ -55,11 +55,22 @@ The fallback identifier is validated only if needed; a working trusted tenant plugin does not require it. Missing both routes, platform credentials, or a working backend fails closed. A user's API key is never substituted. -Moderation dispatches directly to the selected plugin identifier with platform -credentials. Tenants without a trusted OpenAI installation can use the SRE -fallback package. The plugin -runtime must already be installed and loaded in plugin daemon; identifier -dispatch does not bootstrap a missing package. Moderation does not construct a +The verified tenant route uses `/plugin/{tenant_id}/dispatch/moderation/invoke`, +which authorizes the plugin through the tenant's installation record. The SRE +fallback instead uses `/v2/invoke/dispatch/moderation/invoke` with a mandatory +`X-Plugin-Unique-Identifier` and the admitted `tenant_id` in JSON. Both routes +keep daemon server-key authentication, platform credentials, and pinned package +identity. An arbitrary client request cannot select direct dispatch or replace +the server-managed fallback identifier. + +The v2 route is the daemon's existing direct-runtime invocation surface, originally +introduced for Slim remote CLI calls. It parses the supplied identifier, preserves +tenant execution context and cluster routing, and does not query tenant plugin +installations. It is not an installation endpoint or a public tenant API. The +fallback runtime must already be deployed and available to the daemon; direct +dispatch does not bootstrap a missing package. The fallback remains an SRE-owned +trust decision, while tenant-selected packages still require Marketplace and +verified-manifest checks. Moderation does not construct a tenant model instance or enter `CreditPoolService`, so it neither reserves nor deducts user credits. The target LLM retains its normal accounting, including charging consumed usage when output is rejected.