fix(api): use direct dispatch for hosted moderation fallback

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")

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

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
 

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(
             (

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):

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:

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

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.