diff --git a/api/controllers/inner_api/plugin/plugin.py b/api/controllers/inner_api/plugin/plugin.py
index 221887f73c7..bc1f2c4499e 100644
--- a/api/controllers/inner_api/plugin/plugin.py
+++ b/api/controllers/inner_api/plugin/plugin.py
@@ -341,10 +341,12 @@ class PluginInvokeAppApi(Resource):
     )
     @with_session
     def post(self, session: Session, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestInvokeApp):
+        app_user_id = payload.user or getattr(user_model, "session_id", user_model.id)
+
         response = PluginAppBackwardsInvocation.invoke_app(
             session=session,
             app_id=payload.app_id,
-            user_id=user_model.id,
+            user_id=app_user_id,
             tenant_id=tenant_model.id,
             conversation_id=payload.conversation_id,
             query=payload.query,
diff --git a/api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py b/api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py
index 102c4ab4b53..931b1877263 100644
--- a/api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py
+++ b/api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py
@@ -8,6 +8,7 @@ handler tests use inspect.unwrap() to bypass them.
 """
 
 import inspect
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -205,6 +206,78 @@ class TestPluginInvokeAppApi:
         assert hasattr(api_instance, "post")
         assert callable(api_instance.post)
 
+    @patch("controllers.inner_api.plugin.plugin.length_prefixed_response", return_value="stream-response")
+    @patch("controllers.inner_api.plugin.plugin.PluginAppBackwardsInvocation")
+    def test_post_uses_payload_user_for_app_user_id(self, mock_invocation, mock_response, api_instance, app: Flask):
+        mock_payload = SimpleNamespace(
+            app_id="app-1",
+            user="wecom-user",
+            conversation_id=None,
+            query="hello",
+            response_mode="blocking",
+            inputs={"x": 1},
+            files=[],
+        )
+        mock_user = SimpleNamespace(id="wrapper-user-id", session_id="wrapper-session")
+        mock_tenant = SimpleNamespace(id="tenant-1")
+        mock_session = MagicMock()
+        mock_invocation.invoke_app.return_value = {"ok": True}
+        mock_invocation.convert_to_event_stream.return_value = iter([b"{}"])
+
+        raw_post = _extract_raw_post(PluginInvokeAppApi)
+        result = raw_post(
+            api_instance, session=mock_session, user_model=mock_user, tenant_model=mock_tenant, payload=mock_payload
+        )
+
+        assert result == "stream-response"
+        mock_invocation.invoke_app.assert_called_once_with(
+            session=mock_session,
+            app_id="app-1",
+            user_id="wecom-user",
+            tenant_id="tenant-1",
+            conversation_id=None,
+            query="hello",
+            stream=False,
+            inputs={"x": 1},
+            files=[],
+        )
+
+    @patch("controllers.inner_api.plugin.plugin.length_prefixed_response", return_value="stream-response")
+    @patch("controllers.inner_api.plugin.plugin.PluginAppBackwardsInvocation")
+    def test_post_falls_back_to_wrapper_session_id(self, mock_invocation, mock_response, api_instance, app: Flask):
+        mock_payload = SimpleNamespace(
+            app_id="app-1",
+            user=None,
+            conversation_id="conv-1",
+            query="hello",
+            response_mode="streaming",
+            inputs={},
+            files=[],
+        )
+        mock_user = SimpleNamespace(id="wrapper-user-id", session_id="wrapper-session")
+        mock_tenant = SimpleNamespace(id="tenant-1")
+        mock_session = MagicMock()
+        mock_invocation.invoke_app.return_value = {"ok": True}
+        mock_invocation.convert_to_event_stream.return_value = iter([b"{}"])
+
+        raw_post = _extract_raw_post(PluginInvokeAppApi)
+        result = raw_post(
+            api_instance, session=mock_session, user_model=mock_user, tenant_model=mock_tenant, payload=mock_payload
+        )
+
+        assert result == "stream-response"
+        mock_invocation.invoke_app.assert_called_once_with(
+            session=mock_session,
+            app_id="app-1",
+            user_id="wrapper-session",
+            tenant_id="tenant-1",
+            conversation_id="conv-1",
+            query="hello",
+            stream=True,
+            inputs={},
+            files=[],
+        )
+
 
 class TestPluginInvokeEncryptApi:
     """Test PluginInvokeEncryptApi endpoint structure and handler logic"""
diff --git a/api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py b/api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py
index 3716628927b..778b4ba57a3 100644
--- a/api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py
+++ b/api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py
@@ -211,6 +211,33 @@ class TestPluginAppBackwardsInvocation:
         assert route.call_args.args[1] is workflow
         assert route.call_args.args[2] is end_user
 
+    def test_invoke_app_creates_app_scoped_end_user_for_session_user(self, mocker: MockerFixture):
+        app = MagicMock(mode=AppMode.CHAT)
+        end_user = MagicMock(id="end-user", session_id="wecom-user")
+        mocker.patch.object(PluginAppBackwardsInvocation, "_get_app", return_value=app)
+        mocker.patch.object(PluginAppBackwardsInvocation, "_get_user", side_effect=ValueError("user not found"))
+        get_or_create = mocker.patch(
+            "core.plugin.backwards_invocation.app.EndUserService.get_or_create_end_user",
+            return_value=end_user,
+        )
+        route = mocker.patch.object(PluginAppBackwardsInvocation, "invoke_chat_app", return_value={"ok": True})
+
+        result = PluginAppBackwardsInvocation.invoke_app(
+            MagicMock(),
+            app_id="app",
+            user_id="wecom-user",
+            tenant_id="tenant",
+            conversation_id=None,
+            query="hello",
+            stream=False,
+            inputs={},
+            files=[],
+        )
+
+        assert result == {"ok": True}
+        get_or_create.assert_called_once_with(app, user_id="wecom-user")
+        assert route.call_args.args[2] is end_user
+
     def test_invoke_app_missing_query_for_chat_raises(self, mocker: MockerFixture):
         mocker.patch.object(PluginAppBackwardsInvocation, "_get_app", return_value=_app(mode=AppMode.CHAT))
         mocker.patch.object(PluginAppBackwardsInvocation, "_get_user", return_value=_end_user())