diff --git a/api/core/app/apps/pipeline/pipeline_generator.py b/api/core/app/apps/pipeline/pipeline_generator.py index 58f86248ea3..0def3d10ea1 100644 --- a/api/core/app/apps/pipeline/pipeline_generator.py +++ b/api/core/app/apps/pipeline/pipeline_generator.py @@ -24,7 +24,7 @@ from core.app.apps.pipeline.pipeline_queue_manager import PipelineQueueManager from core.app.apps.pipeline.pipeline_runner import PipelineDocumentStore, PipelineRunner from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline -from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity, UserFrom from core.app.entities.rag_pipeline_invoke_entities import RagPipelineInvokeEntity from core.app.entities.task_entities import ( WorkflowAppBlockingResponse, @@ -197,6 +197,7 @@ class PipelineGenerator(BaseAppGenerator): ) session.add(document_pipeline_execution_log) application_generate_entity = RagPipelineGenerateEntity( + user_from=UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER, task_id=str(uuid.uuid4()), app_config=pipeline_config, pipeline_config=pipeline_config, @@ -428,6 +429,7 @@ class PipelineGenerator(BaseAppGenerator): # init application generate entity - use RagPipelineGenerateEntity instead application_generate_entity = RagPipelineGenerateEntity( + user_from=UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER, task_id=str(uuid.uuid4()), app_config=pipeline_config, pipeline_config=pipeline_config, @@ -530,6 +532,7 @@ class PipelineGenerator(BaseAppGenerator): # init application generate entity application_generate_entity = RagPipelineGenerateEntity( + user_from=UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER, task_id=str(uuid.uuid4()), app_config=pipeline_config, pipeline_config=pipeline_config, diff --git a/api/core/app/apps/workflow/app_generator.py b/api/core/app/apps/workflow/app_generator.py index 806a4e76397..daf053c8b4d 100644 --- a/api/core/app/apps/workflow/app_generator.py +++ b/api/core/app/apps/workflow/app_generator.py @@ -25,7 +25,7 @@ from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager from core.app.apps.workflow.app_runner import WorkflowAppRunner from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.entities.task_entities import ( WorkflowAppBlockingResponse, WorkflowAppPausedBlockingResponse, @@ -55,6 +55,7 @@ from models.account import Account from models.enums import WorkflowRunTriggeredFrom from models.model import App, EndUser from models.workflow import Workflow, WorkflowNodeExecutionTriggeredFrom +from repositories.entities.workflow_execution import WorkflowExecutionIdentity from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService if TYPE_CHECKING: @@ -97,6 +98,7 @@ class WorkflowAppGenerator(BaseAppGenerator): def generate( self, *, + execution_identity: WorkflowExecutionIdentity, app_model: App, workflow: Workflow, user: Account | EndUser, @@ -116,6 +118,7 @@ class WorkflowAppGenerator(BaseAppGenerator): def generate( self, *, + execution_identity: WorkflowExecutionIdentity, app_model: App, workflow: Workflow, user: Account | EndUser, @@ -135,6 +138,7 @@ class WorkflowAppGenerator(BaseAppGenerator): def generate( self, *, + execution_identity: WorkflowExecutionIdentity, app_model: App, workflow: Workflow, user: Account | EndUser, @@ -153,6 +157,7 @@ class WorkflowAppGenerator(BaseAppGenerator): def generate( self, *, + execution_identity: WorkflowExecutionIdentity, app_model: App, workflow: Workflow, user: Account | EndUser, @@ -223,7 +228,8 @@ class WorkflowAppGenerator(BaseAppGenerator): file_upload_config=file_extra_config, inputs=inputs, files=list(system_files), - user_id=user.id, + user_id=execution_identity.user_id, + user_from=execution_identity.user_from, stream=streaming, invoke_from=invoke_from, allow_human_input=allow_human_input, @@ -432,6 +438,7 @@ class WorkflowAppGenerator(BaseAppGenerator): args: Mapping[str, Any], streaming: bool = True, *, + execution_identity: WorkflowExecutionIdentity, session: Session, ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any]]: """ @@ -460,7 +467,8 @@ class WorkflowAppGenerator(BaseAppGenerator): app_config=app_config, inputs={}, files=[], - user_id=user.id, + user_id=execution_identity.user_id, + user_from=execution_identity.user_from, stream=streaming, invoke_from=InvokeFrom.DEBUGGER, extras={ @@ -526,6 +534,7 @@ class WorkflowAppGenerator(BaseAppGenerator): args: LoopNodeRunPayload, streaming: bool = True, *, + execution_identity: WorkflowExecutionIdentity, session: Session, ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any]]: """ @@ -554,7 +563,8 @@ class WorkflowAppGenerator(BaseAppGenerator): app_config=app_config, inputs={}, files=[], - user_id=user.id, + user_id=execution_identity.user_id, + user_from=execution_identity.user_from, stream=streaming, invoke_from=InvokeFrom.DEBUGGER, extras={ @@ -651,18 +661,16 @@ class WorkflowAppGenerator(BaseAppGenerator): workflow_run_id=application_generate_entity.workflow_execution_id, ) - # Determine system_user_id based on invocation source - is_external_api_call = application_generate_entity.invoke_from in { - InvokeFrom.WEB_APP, - InvokeFrom.SERVICE_API, - } - - if is_external_api_call: - # For external API calls, use end user's session ID - end_user = session.scalar(select(EndUser).where(EndUser.id == application_generate_entity.user_id)) + user_from = application_generate_entity.user_from + if user_from == UserFrom.END_USER: + end_user = session.scalar( + select(EndUser).where( + EndUser.tenant_id == application_generate_entity.app_config.tenant_id, + EndUser.id == application_generate_entity.user_id, + ) + ) system_user_id = end_user.session_id if end_user else "" else: - # For internal calls, use the original user ID system_user_id = application_generate_entity.user_id runner = WorkflowAppRunner( diff --git a/api/core/app/apps/workflow/app_runner.py b/api/core/app/apps/workflow/app_runner.py index 5752e14b987..b2390ea19b1 100644 --- a/api/core/app/apps/workflow/app_runner.py +++ b/api/core/app/apps/workflow/app_runner.py @@ -79,7 +79,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner): # if only single iteration or single loop run is requested if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run: invoke_from = InvokeFrom.DEBUGGER - user_from = self._resolve_user_from(invoke_from) + user_from = self.application_generate_entity.user_from resume_state = self._resume_graph_runtime_state diff --git a/api/core/app/entities/app_invoke_entities.py b/api/core/app/entities/app_invoke_entities.py index 750124ab6c0..6f0baa4a586 100644 --- a/api/core/app/entities/app_invoke_entities.py +++ b/api/core/app/entities/app_invoke_entities.py @@ -62,8 +62,9 @@ class InvokeFrom(StrEnum): """Whether a run from this entry point is attributed to a workspace Account rather than an end user. Console contexts (debugger/explore) run as the signed-in Account; webapp/service-api/trigger run as an - EndUser. Single source of truth for the created-by-role / user-type - split shared by the app runners and MCP identity forwarding.""" + EndUser by default. Explicit caller identity takes precedence: scheduled + workflows can use SERVICE_API with an Account. This inference is shared + by app runners without caller metadata and MCP identity forwarding.""" return self in (InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE) @@ -342,6 +343,8 @@ class WorkflowAppGenerateEntity(AppGenerateEntity): # app config app_config: WorkflowUIBasedAppConfig = None # type: ignore + # Every new execution supplies its actual role. Legacy snapshots are normalized when loaded. + user_from: UserFrom workflow_execution_id: str workflow_execution_timing: WorkflowExecutionTiming = Field( default_factory=WorkflowExecutionTiming, exclude=True, repr=False diff --git a/api/core/app/entities/rag_pipeline_invoke_entities.py b/api/core/app/entities/rag_pipeline_invoke_entities.py index 992b8da8936..2269f1cbc9a 100644 --- a/api/core/app/entities/rag_pipeline_invoke_entities.py +++ b/api/core/app/entities/rag_pipeline_invoke_entities.py @@ -1,6 +1,8 @@ from typing import Any -from pydantic import BaseModel +from pydantic import BaseModel, field_validator + +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom class RagPipelineInvokeEntity(BaseModel): @@ -12,3 +14,19 @@ class RagPipelineInvokeEntity(BaseModel): streaming: bool workflow_execution_id: str | None = None workflow_thread_pool_id: str | None = None + + @field_validator("application_generate_entity") + @classmethod + def normalize_legacy_execution_identity(cls, value: dict[str, Any]) -> dict[str, Any]: + """Load queued payloads written before user_from became required. + + Keep this compatibility at the serialized queue boundary, rather than + weakening validation of new execution entities or changing runner policy. + """ + if "user_from" in value or "invoke_from" not in value: + return value + invoke_from = InvokeFrom(value["invoke_from"]) + if value.get("single_iteration_run") or value.get("single_loop_run"): + invoke_from = InvokeFrom.DEBUGGER + user_from = UserFrom.ACCOUNT if invoke_from.runs_as_account() else UserFrom.END_USER + return {**value, "user_from": user_from.value} diff --git a/api/core/app/layers/pause_state_persist_layer.py b/api/core/app/layers/pause_state_persist_layer.py index a228611bc59..497cd65ed08 100644 --- a/api/core/app/layers/pause_state_persist_layer.py +++ b/api/core/app/layers/pause_state_persist_layer.py @@ -1,11 +1,16 @@ from dataclasses import dataclass -from typing import Annotated, Literal, Self +from typing import Annotated, Any, Literal, Self -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator from sqlalchemy import Engine from sqlalchemy.orm import Session, sessionmaker -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from graphon.engine.filter import ResponseStreamFilter from models.model import AppMode from repositories.entities.workflow_execution import WorkflowExecutionTiming @@ -18,6 +23,19 @@ class _WorkflowGenerateEntityWrapper(BaseModel): type: Literal[AppMode.WORKFLOW] = AppMode.WORKFLOW entity: WorkflowAppGenerateEntity + @model_validator(mode="before") + @classmethod + def normalize_legacy_identity(cls, value: Any) -> Any: + """Preserve the previous runner role for snapshots written before explicit identity.""" + if isinstance(value, dict) and isinstance(entity := value.get("entity"), dict) and "user_from" not in entity: + invoke_from = InvokeFrom(entity["invoke_from"]) + # Single-node debugging previously forced DEBUGGER before resolving the role. + if entity.get("single_iteration_run") or entity.get("single_loop_run"): + invoke_from = InvokeFrom.DEBUGGER + user_from = UserFrom.ACCOUNT if invoke_from.runs_as_account() else UserFrom.END_USER + return {**value, "entity": {**entity, "user_from": user_from}} + return value + class _AdvancedChatAppGenerateEntityWrapper(BaseModel): type: Literal[AppMode.ADVANCED_CHAT] = AppMode.ADVANCED_CHAT diff --git a/api/core/plugin/backwards_invocation/app.py b/api/core/plugin/backwards_invocation/app.py index 7a12c5a0be6..b909d994bbd 100644 --- a/api/core/plugin/backwards_invocation/app.py +++ b/api/core/plugin/backwards_invocation/app.py @@ -26,6 +26,7 @@ from models.model import ( load_annotation_reply_config, ) from models.workflow import Workflow +from services.workflow_execution_identity import build_workflow_execution_identity if TYPE_CHECKING: from core.app.apps.workflow_app_runner import WorkflowRunDriver @@ -214,6 +215,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation): app_model=app, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args={"inputs": inputs, "files": files}, invoke_from=InvokeFrom.SERVICE_API, streaming=stream, diff --git a/api/core/tools/workflow_as_tool/tool.py b/api/core/tools/workflow_as_tool/tool.py index dc46263692a..9598b4ebb64 100644 --- a/api/core/tools/workflow_as_tool/tool.py +++ b/api/core/tools/workflow_as_tool/tool.py @@ -34,6 +34,7 @@ from models import Account, Tenant from models.model import App, EndUser from models.utils.file_input_compat import build_file_from_stored_mapping from models.workflow import Workflow +from services.workflow_execution_identity import build_workflow_execution_identity if TYPE_CHECKING: from core.app.apps.workflow_app_runner import WorkflowRunDriver @@ -129,6 +130,7 @@ class WorkflowTool(Tool): app_model=app, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=generator_args, invoke_from=self.runtime.invoke_from, streaming=False, diff --git a/api/repositories/entities/workflow_execution.py b/api/repositories/entities/workflow_execution.py index 905b2cfb5d6..80b964b5b4e 100644 --- a/api/repositories/entities/workflow_execution.py +++ b/api/repositories/entities/workflow_execution.py @@ -1,14 +1,27 @@ """Workflow execution values shared across runtime and storage boundaries.""" +from __future__ import annotations + from dataclasses import dataclass, field from threading import Lock from time import perf_counter -from typing import override +from typing import TYPE_CHECKING, override from pydantic import Field from graphon.entities import WorkflowExecution as GraphonWorkflowExecution +if TYPE_CHECKING: + from core.app.entities.app_invoke_entities import UserFrom + + +@dataclass(frozen=True) +class WorkflowExecutionIdentity: + """Resolved caller values crossing execution boundaries without ORM objects.""" + + user_id: str + user_from: UserFrom + @dataclass class WorkflowExecutionTiming: diff --git a/api/services/app_generate_service.py b/api/services/app_generate_service.py index 48e6d50ae1f..0dc5d8297ca 100644 --- a/api/services/app_generate_service.py +++ b/api/services/app_generate_service.py @@ -33,6 +33,7 @@ from services.errors.app import ( ) from services.errors.llm import InvokeRateLimitError from services.quota_service import QuotaService, unlimited +from services.workflow_execution_identity import build_workflow_execution_identity from services.workflow_run_agg import WorkflowRunAgg from services.workflow_service import WorkflowService from tasks.app_generate.workflow_execute_task import AppExecutionParams, workflow_based_app_execution_task @@ -345,6 +346,7 @@ class AppGenerateService: app_model=app_model, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, invoke_from=invoke_from, streaming=False, @@ -419,6 +421,7 @@ class AppGenerateService: workflow=workflow, node_id=node_id, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, streaming=streaming, session=session, @@ -464,6 +467,7 @@ class AppGenerateService: workflow=workflow, node_id=node_id, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, streaming=streaming, session=session, diff --git a/api/services/snippet_generate_service.py b/api/services/snippet_generate_service.py index 2b31cbce4bf..bab575c0a99 100644 --- a/api/services/snippet_generate_service.py +++ b/api/services/snippet_generate_service.py @@ -38,6 +38,7 @@ from models.model import App, AppMode, EndUser from models.snippet import CustomizedSnippet from models.workflow import Workflow, WorkflowNodeExecutionModel from services.snippet_service import SnippetService +from services.workflow_execution_identity import build_workflow_execution_identity from services.workflow_run_agg import WorkflowRunAgg from services.workflow_service import WorkflowService @@ -173,6 +174,7 @@ class SnippetGenerateService: app_model=app_proxy, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, invoke_from=invoke_from, streaming=streaming, @@ -218,6 +220,7 @@ class SnippetGenerateService: app_model=app_proxy, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, invoke_from=invoke_from, streaming=False, @@ -416,6 +419,7 @@ class SnippetGenerateService: workflow=workflow, node_id=node_id, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, streaming=streaming, session=session, @@ -463,6 +467,7 @@ class SnippetGenerateService: workflow=workflow, node_id=node_id, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, # type: ignore[arg-type] streaming=streaming, session=session, diff --git a/api/services/workflow_execution_identity.py b/api/services/workflow_execution_identity.py new file mode 100644 index 00000000000..4b16b26ec1f --- /dev/null +++ b/api/services/workflow_execution_identity.py @@ -0,0 +1,22 @@ +"""Trusted caller identity passed into workflow execution, independent of ORM state.""" + +from typing import TYPE_CHECKING + +from core.app.entities.app_invoke_entities import UserFrom +from repositories.entities.workflow_execution import WorkflowExecutionIdentity + +if TYPE_CHECKING: + from models.account import Account + from models.model import EndUser + + +def build_workflow_execution_identity(user: "Account | EndUser") -> WorkflowExecutionIdentity: + """Adapt a resolved caller at the service/task boundary; never infer a role from the entry point.""" + from models.account import Account + from models.model import EndUser + + if isinstance(user, Account): + return WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.ACCOUNT) + if isinstance(user, EndUser): + return WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER) + raise TypeError("Workflow caller must be an Account or EndUser") diff --git a/api/tasks/app_generate/workflow_execute_task.py b/api/tasks/app_generate/workflow_execute_task.py index 33357466d34..12eec51ec54 100644 --- a/api/tasks/app_generate/workflow_execute_task.py +++ b/api/tasks/app_generate/workflow_execute_task.py @@ -39,6 +39,7 @@ from models.workflow import Workflow, WorkflowNodeExecutionTriggeredFrom, Workfl from repositories.factory import DifyAPIRepositoryFactory from repositories.sqlalchemy_api_workflow_run_repository import WorkflowRunNotPausedError from services.app_task_service import AppTaskService +from services.workflow_execution_identity import build_workflow_execution_identity from services.workflow_run_agg import WorkflowRunAgg logger = logging.getLogger(__name__) @@ -231,6 +232,7 @@ class _AppRunner: app_model=app, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=exec_params.args, invoke_from=exec_params.invoke_from, streaming=exec_params.streaming, diff --git a/api/tasks/async_workflow_tasks.py b/api/tasks/async_workflow_tasks.py index 7e0cf0e14b2..4620f3e55bd 100644 --- a/api/tasks/async_workflow_tasks.py +++ b/api/tasks/async_workflow_tasks.py @@ -39,6 +39,7 @@ from services.workflow.entities import ( WorkflowResumeTaskData, WorkflowTaskData, ) +from services.workflow_execution_identity import build_workflow_execution_identity from services.workflow_run_agg import WorkflowRunAgg from tasks.workflow_cfs_scheduler.cfs_scheduler import AsyncWorkflowCFSPlanEntity, AsyncWorkflowCFSPlanScheduler from tasks.workflow_cfs_scheduler.entities import AsyncWorkflowQueue, AsyncWorkflowSystemStrategy @@ -168,6 +169,7 @@ def _execute_workflow_common( app_model=app_model, workflow=workflow, user=user, + execution_identity=build_workflow_execution_identity(user), args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=False, diff --git a/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py b/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py index 160c4c67039..8e0e4f40ecf 100644 --- a/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py +++ b/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py @@ -10,7 +10,7 @@ from sqlalchemy import Engine from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext, _WorkflowGenerateEntityWrapper from core.workflow.human_input_adapter import DeliveryMethodType from core.workflow.nodes.human_input.entities import ( @@ -98,6 +98,7 @@ def _build_resumption_context(*, app: App, workflow_run: WorkflowRun, options: l workflow_id=workflow_run.workflow_id, ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task-1", app_config=app_config, inputs={}, diff --git a/api/tests/e2e_tests/containers/core/app/layers/test_pause_state_persist_layer.py b/api/tests/e2e_tests/containers/core/app/layers/test_pause_state_persist_layer.py index 7f89508a45d..21941ec3ab4 100644 --- a/api/tests/e2e_tests/containers/core/app/layers/test_pause_state_persist_layer.py +++ b/api/tests/e2e_tests/containers/core/app/layers/test_pause_state_persist_layer.py @@ -55,6 +55,7 @@ class WorkflowCase: pause: bool = True, ) -> tuple[PreparedWorkflowRun, WorkflowRunAgg]: entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=str(uuid4()), app_config=WorkflowUIBasedAppConfig( tenant_id=self.workflow.tenant_id, diff --git a/api/tests/e2e_tests/containers/core/workflow/test_human_input_resume_node_execution.py b/api/tests/e2e_tests/containers/core/workflow/test_human_input_resume_node_execution.py index 2c8e465c5f3..926c860df5d 100644 --- a/api/tests/e2e_tests/containers/core/workflow/test_human_input_resume_node_execution.py +++ b/api/tests/e2e_tests/containers/core/workflow/test_human_input_resume_node_execution.py @@ -6,7 +6,7 @@ from sqlalchemy import delete, select from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.repositories.human_input_repository import HumanInputFormRepository, HumanInputFormRepositoryImpl from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository @@ -137,6 +137,7 @@ def _build_generate_entity( workflow_id=workflow_id, ) return WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=str(uuid.uuid4()), app_config=app_config, inputs={}, diff --git a/api/tests/e2e_tests/containers/services/workflow/test_workflow_event_snapshot_service.py b/api/tests/e2e_tests/containers/services/workflow/test_workflow_event_snapshot_service.py index 0e7a92c9225..ca8d6e294c4 100644 --- a/api/tests/e2e_tests/containers/services/workflow/test_workflow_event_snapshot_service.py +++ b/api/tests/e2e_tests/containers/services/workflow/test_workflow_event_snapshot_service.py @@ -10,7 +10,7 @@ from sqlalchemy import Engine, delete from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext, _WorkflowGenerateEntityWrapper from core.workflow.nodes.human_input.entities import SelectInputConfig, StringListSource, UserActionConfig from core.workflow.nodes.human_input.enums import HumanInputFormStatus, ValueSourceType @@ -69,6 +69,7 @@ def _build_resumption_context(workflow_run_id: str) -> WorkflowResumptionContext workflow_id=str(uuid4()), ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task-1", app_config=app_config, inputs={}, diff --git a/api/tests/e2e_tests/containers/tasks/test_mail_human_input_delivery_task.py b/api/tests/e2e_tests/containers/tasks/test_mail_human_input_delivery_task.py index 6f5b20c4683..32b2a342109 100644 --- a/api/tests/e2e_tests/containers/tasks/test_mail_human_input_delivery_task.py +++ b/api/tests/e2e_tests/containers/tasks/test_mail_human_input_delivery_task.py @@ -8,7 +8,7 @@ from sqlalchemy.orm import Session from configs import dify_config from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext from core.repositories.human_input_repository import FormCreateParams, HumanInputFormRepositoryImpl from core.workflow.human_input_adapter import ( @@ -143,6 +143,7 @@ def _create_workflow_pause_state( generate_entity={ "type": AppMode.WORKFLOW, "entity": WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id=str(uuid.uuid4()), app_config=WorkflowUIBasedAppConfig( tenant_id=tenant_id, diff --git a/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py b/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py index e4356ea2064..a55bb2b0ed0 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py @@ -23,7 +23,12 @@ from core.app.app_config.entities import AppAdditionalFeatures, WorkflowUIBasedA from core.app.apps.advanced_chat.generate_task_pipeline import ConversationSnapshot, MessageSnapshot, WorkflowSnapshot from core.app.apps.common import workflow_response_converter from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from core.app.entities.queue_entities import QueueWorkflowPausedEvent from core.app.entities.task_entities import ( AdvancedChatPausedBlockingResponse, @@ -158,6 +163,7 @@ def _persist_human_input_form( def _build_service_api_pause_converter() -> WorkflowResponseConverter: application_generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task", workflow_execution_id="run-id", user_id="user", @@ -317,6 +323,7 @@ def _build_resumption_context(task_id: str) -> WorkflowResumptionContext: workflow_id="workflow-1", ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=task_id, app_config=app_config, inputs={}, diff --git a/api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter.py b/api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter.py index c80352323d3..2b7753b3d3b 100644 --- a/api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter.py +++ b/api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter.py @@ -14,7 +14,7 @@ from core.app.app_config.entities import WorkflowUIBasedAppConfig from core.app.apps.common import workflow_response_converter from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter from core.app.apps.workflow.app_runner import WorkflowAppRunner -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.entities.queue_entities import ( NodeExecutionSnapshot, QueueEvent, @@ -320,6 +320,7 @@ def _workflow_response_converter_human_input_build_converter() -> WorkflowRespon workflow_execution_id="run-1", ) app_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task-1", app_config=WorkflowUIBasedAppConfig( app_id="app-1", tenant_id="tenant-1", workflow_id="wf-1", app_mode=AppMode.WORKFLOW @@ -439,6 +440,7 @@ def _workflow_response_converter_resumption_build_converter() -> WorkflowRespons workflow_execution_id="run-1", ) app_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task-1", app_config=WorkflowUIBasedAppConfig( app_id="app-1", tenant_id="tenant-1", workflow_id="wf-1", app_mode=AppMode.WORKFLOW @@ -626,6 +628,7 @@ class TestConverterTruncationCases: def create_generate_entity(self) -> WorkflowAppGenerateEntity: """Create a validated workflow invocation for response conversion.""" return WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, app_config=WorkflowUIBasedAppConfig( tenant_id="test-tenant-id", app_id="app", workflow_id="wf-id", app_mode=AppMode.WORKFLOW ), @@ -1010,6 +1013,7 @@ class TestConverterTruncationCases: ) entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="test_task_id", app_id="test_app_id", app_config=app_config, @@ -1548,6 +1552,7 @@ def _workflow_pause_events_build_converter( *, invoke_from: InvokeFrom = InvokeFrom.SERVICE_API ) -> WorkflowResponseConverter: application_generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task", workflow_execution_id="run-id", user_id="user", diff --git a/api/tests/unit_tests/core/app/apps/pipeline/test_pipeline_runner.py b/api/tests/unit_tests/core/app/apps/pipeline/test_pipeline_runner.py index 35637ce277a..4c7e8a1fbad 100644 --- a/api/tests/unit_tests/core/app/apps/pipeline/test_pipeline_runner.py +++ b/api/tests/unit_tests/core/app/apps/pipeline/test_pipeline_runner.py @@ -23,7 +23,7 @@ import core.app.apps.pipeline.pipeline_runner as module from core.app.apps.pipeline.pipeline_config_manager import PipelineConfig from core.app.apps.pipeline.pipeline_queue_manager import PipelineQueueManager from core.app.apps.pipeline.pipeline_runner import PipelineRunner -from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity, UserFrom from core.app.entities.queue_entities import AppQueueEvent, QueueWorkflowFailedEvent from core.workflow.workflow_entry import WorkflowEntry from graphon.engine_events import EngineEvent, GraphRunFailedEvent @@ -102,6 +102,7 @@ def _persist_scope( def _build_app_generate_entity() -> RagPipelineGenerateEntity: app_config = PipelineConfig(app_id="pipe", workflow_id="wf", tenant_id="tenant", app_mode=AppMode.RAG_PIPELINE) return RagPipelineGenerateEntity( + user_from=UserFrom.END_USER, task_id="task", stream=True, app_config=app_config, diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner.py index 5768e5f3d45..cd00141bb80 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner.py +++ b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner.py @@ -1116,6 +1116,7 @@ class TestWorkflowAppRunnerSingleNodeCases: queue_manager: WorkflowAppQueueManager, ) -> None: app_generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, app_config=WorkflowUIBasedAppConfig( app_id="app", tenant_id="tenant", workflow_id="workflow", app_mode=AppMode.WORKFLOW ), @@ -1268,6 +1269,7 @@ class TestWorkflowAppRunnerSingleNodeCases: @staticmethod def test_run_adds_inputs_with_snippet_compatible_start_aliases(queue_manager: WorkflowAppQueueManager) -> None: app_generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, app_config=WorkflowUIBasedAppConfig( app_id="app", tenant_id="tenant", workflow_id="workflow", app_mode=AppMode.WORKFLOW ), @@ -1426,6 +1428,7 @@ def _make_paused_workflow( def _resume_events(runner: WorkflowBasedAppRunner, entry: WorkflowEntry) -> Generator[EngineEvent]: """Resume through orchestration with actual repositories and the real Engine.""" entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task", app_config=WorkflowUIBasedAppConfig( tenant_id="tenant", diff --git a/api/tests/unit_tests/core/app/apps/workflow/test_app_generator.py b/api/tests/unit_tests/core/app/apps/workflow/test_app_generator.py index 98b8d0411ae..16fc32e9b47 100644 --- a/api/tests/unit_tests/core/app/apps/workflow/test_app_generator.py +++ b/api/tests/unit_tests/core/app/apps/workflow/test_app_generator.py @@ -12,17 +12,28 @@ from sqlalchemy import inspect from sqlalchemy.orm import Session, sessionmaker import core.app.apps.workflow.app_generator as _workflow_app_generator +from controllers.inner_api.plugin.wraps import get_user from core.app.app_config.entities import AppAdditionalFeatures, WorkflowUIBasedAppConfig from core.app.apps.draft_variable_saver import DraftVariableSaverFactory from core.app.apps.exc import GenerateTaskStoppedError from core.app.apps.workflow import app_generator as _app_generator_extra_app_generator_module from core.app.apps.workflow.app_generator import WorkflowAppGenerator from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.entities.queue_entities import QueueWorkflowSucceededEvent -from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig +from core.app.layers.pause_state_persist_layer import ( + PauseStateLayerConfig, + WorkflowResumptionContext, + _WorkflowGenerateEntityWrapper, +) from core.ops.ops_trace_manager import OpsTraceManager, TraceQueueManager from core.repositories import SQLAlchemyWorkflowExecutionRepository, SQLAlchemyWorkflowNodeExecutionRepository +from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository +from core.tools.tool_file_manager import ToolFileManager +from core.workflow.file_reference import build_file_reference +from core.workflow.node_runtime import resolve_dify_run_context +from core.workflow.system_variables import SystemVariableKey, get_system_text from graphon.enums import WorkflowExecutionStatus from graphon.runtime import RuntimeState, VariablePool from graphon.variable_loader import DUMMY_VARIABLE_LOADER @@ -30,8 +41,10 @@ from models.enums import CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom from models.model import App, AppMode, EndUser from models.snippet import CustomizedSnippet from models.workflow import Workflow, WorkflowKind, WorkflowNodeExecutionTriggeredFrom, WorkflowRun, WorkflowType +from repositories.entities.workflow_execution import WorkflowExecutionIdentity +from services.file_request_service import FileRequestService from services.workflow_run_agg import WorkflowRunAgg -from tests.helpers.unit.model_factories import make_workflow +from tests.helpers.unit.model_factories import make_account, make_workflow """SQLite-backed tests for workflow app generation and worker reload behavior.""" @@ -98,6 +111,7 @@ def _generate_entity( stream: bool = True, ) -> WorkflowAppGenerateEntity: return WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task", app_config=_app_config(app, workflow), inputs={}, @@ -167,6 +181,7 @@ class TestWorkflowAppGeneratorCases: ) sqlite_session.commit() entity = WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="worker-task", app_config=_app_config(app_model, workflow), inputs={}, @@ -312,6 +327,7 @@ class TestWorkflowAppGeneratorCases: app_model=app, workflow=workflow, user=end_user, + execution_identity=WorkflowExecutionIdentity(user_id=end_user.id, user_from=UserFrom.ACCOUNT), args={ "inputs": {"query": "hello"}, "files": [], @@ -718,6 +734,138 @@ class TestAppGeneratorExtraCases: ) return sqlite_session + @pytest.mark.parametrize("resuming", ["fresh", "new-snapshot", "legacy-snapshot"]) + @pytest.mark.parametrize("is_account", [True, False], ids=["account", "end-user"]) + def test_service_api_preserves_caller_identity_and_file_access( + self, + monkeypatch: pytest.MonkeyPatch, + sqlite_generator_session: Session, + sqlite_session_factory: sessionmaker[Session], + resuming: str, + is_account: bool, + ) -> None: + app_model, workflow, end_user = _persist_generator_rows(sqlite_generator_session) + account = make_account(account_id="scheduled-account") + user = account if is_account else end_user + expected_user_from = UserFrom.ACCOUNT if is_account and resuming != "legacy-snapshot" else UserFrom.END_USER + expected_system_user_id = account.id if is_account else end_user.session_id + entity = _generate_entity(app_model, workflow, end_user, stream=False) + entity.user_id = user.id + entity.user_from = expected_user_from + runtime_state = None + if resuming != "fresh": + # New paused runs retain the explicit identity across serialization. + payload = WorkflowResumptionContext( + generate_entity=_WorkflowGenerateEntityWrapper(entity=entity), + serialized_graph_runtime_state="", + ).model_dump(mode="json") + if resuming == "legacy-snapshot": + payload["generate_entity"]["entity"].pop("user_from") + restored = WorkflowResumptionContext.loads(json.dumps(payload)).get_generate_entity() + assert isinstance(restored, WorkflowAppGenerateEntity) + entity = restored + runtime_state = _runtime_state() + runtime_state.variable_pool.add(["sys", "user_id"], expected_system_user_id) + execution_repository, node_repository = _repositories(sqlite_session_factory, app_model, end_user) + observed: list[tuple[str, UserFrom, str]] = [] + + def inspect_execution( + runner: WorkflowBasedAppRunner, + _pause_config: PauseStateLayerConfig | None, + _executions: WorkflowExecutionRepository, + _nodes: WorkflowNodeExecutionRepository, + ) -> None: + prepared = runner.prepare() + assert prepared is not None + context = resolve_dify_run_context(prepared.entry.graph_engine.graph.root_node.run_context) + system_user_id = get_system_text( + prepared.entry.graph_engine.runtime_state.variable_pool, SystemVariableKey.USER_ID + ) + assert context.user_from == expected_user_from + # Old scheduled snapshots retain the Agent upload's EndUser remapping, + # including its known mismatch with download's original Account id. + legacy_scheduled = resuming == "legacy-snapshot" and is_account + owner_id = get_user(context.tenant_id, context.user_id).id if legacy_scheduled else context.user_id + tool_file = ToolFileManager().create_file_by_raw( + user_id=owner_id, + tenant_id=context.tenant_id, + conversation_id=None, + file_binary=b"scheduled report", + mimetype="text/plain", + filename="report.txt", + ) + mapping = {"transfer_method": "tool_file", "reference": build_file_reference(record_id=tool_file.id)} + service = FileRequestService() + if legacy_scheduled: + assert owner_id != context.user_id + with pytest.raises(ValueError, match="not found"): + service.request_download( + tenant_id=context.tenant_id, + user_id=context.user_id, + user_from=context.user_from, + invoke_from=context.invoke_from, + file_mapping=mapping, + ) + observed.append((context.user_id, context.user_from, system_user_id or "")) + return + result = service.request_download( + tenant_id=context.tenant_id, + user_id=context.user_id, + user_from=context.user_from, + invoke_from=context.invoke_from, + file_mapping=mapping, + ) + assert result.filename == "report.txt" + assert result.size == len(b"scheduled report") + assert tool_file.id in result.download_uri + for tenant_id, user_id, user_from in [ + (context.tenant_id, "another-end-user", UserFrom.END_USER), + ("another-tenant", context.user_id, context.user_from), + ]: + with pytest.raises(ValueError, match="not found"): + service.request_download( + tenant_id=tenant_id, + user_id=user_id, + user_from=user_from, + invoke_from=InvokeFrom.SERVICE_API, + file_mapping=mapping, + ) + + observed.append((context.user_id, context.user_from, system_user_id or "")) + + generator = WorkflowAppGenerator(execution_driver=inspect_execution) + monkeypatch.setattr(generator, "_handle_response", lambda **kwargs: {}) + monkeypatch.setattr( + _workflow_app_generator.WorkflowAppGenerateResponseConverter, + "convert", + lambda response, invoke_from: response, + ) + monkeypatch.setattr(generator, "_restore_workflow_run_graph", lambda **kwargs: None) + monkeypatch.setattr("core.tools.tool_file_manager.storage.save", lambda *args: None) + monkeypatch.setattr("controllers.inner_api.plugin.wraps.db", _workflow_app_generator.db) + if resuming == "fresh": + generator.generate( + app_model=app_model, + workflow=workflow, + user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=expected_user_from), + args={"inputs": {}}, + invoke_from=InvokeFrom.SERVICE_API, + streaming=False, + ) + else: + assert runtime_state is not None + generator.resume( + app_model=app_model, + workflow=workflow, + user=user, + application_generate_entity=entity, + workflow_execution_repository=execution_repository, + workflow_node_execution_repository=node_repository, + graph_runtime_state=runtime_state, + ) + assert observed == [(user.id, expected_user_from, expected_system_user_id)] + class TestWorkflowAppGeneratorValidation: @pytest.mark.usefixtures("sqlite_generator_session") def test_generate_stream_joins_worker_after_response_exhaustion( @@ -738,6 +886,7 @@ class TestAppGeneratorExtraCases: workflow_id="workflow-id", ) application_generate_entity = WorkflowAppGenerateEntity.model_construct( + user_from=UserFrom.END_USER, task_id="task", app_config=app_config, inputs={}, @@ -843,6 +992,7 @@ class TestAppGeneratorExtraCases: workflow=_app_generator_extra_workflow(), node_id="", user=_end_user(), + execution_identity=WorkflowExecutionIdentity(user_id=_end_user().id, user_from=UserFrom.END_USER), args={"inputs": {}}, streaming=False, session=sqlite_session, @@ -854,6 +1004,7 @@ class TestAppGeneratorExtraCases: workflow=_app_generator_extra_workflow(), node_id="node", user=_end_user(), + execution_identity=WorkflowExecutionIdentity(user_id=_end_user().id, user_from=UserFrom.END_USER), args={}, streaming=False, session=sqlite_session, @@ -868,6 +1019,7 @@ class TestAppGeneratorExtraCases: workflow=_app_generator_extra_workflow(), node_id="", user=_end_user(), + execution_identity=WorkflowExecutionIdentity(user_id=_end_user().id, user_from=UserFrom.END_USER), args=SimpleNamespace(inputs={}), streaming=False, session=sqlite_session, @@ -924,6 +1076,7 @@ class TestAppGeneratorExtraCases: workflow=workflow, node_id="node-1", user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER), args={"inputs": {"foo": "bar"}, "trace_session_id": "session-1"}, streaming=False, session=sqlite_session, @@ -986,6 +1139,7 @@ class TestAppGeneratorExtraCases: workflow=workflow, node_id="node-2", user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER), args=SimpleNamespace(inputs={"foo": "bar"}, trace_session_id="session-1"), streaming=False, session=sqlite_session, @@ -1003,6 +1157,7 @@ class TestAppGeneratorExtraCases: workflow=_app_generator_extra_workflow(), node_id="node", user=_end_user(), + execution_identity=WorkflowExecutionIdentity(user_id=_end_user().id, user_from=UserFrom.END_USER), args=SimpleNamespace(inputs=None), streaming=False, session=sqlite_session, @@ -1021,6 +1176,7 @@ class TestAppGeneratorExtraCases: workflow_id="workflow-id", ) application_generate_entity = WorkflowAppGenerateEntity.model_construct( + user_from=UserFrom.END_USER, task_id="task", app_config=app_config, inputs={}, @@ -1088,6 +1244,7 @@ class TestAppGeneratorExtraCases: app_model=app, workflow=workflow, user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER), args={"inputs": inputs}, invoke_from=invoke_from, streaming=False, @@ -1129,6 +1286,7 @@ class TestAppGeneratorExtraCases: app_model=app, workflow=workflow, user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER), args={"inputs": {"question": "hello", "unknown": "filtered"}}, invoke_from=InvokeFrom.SERVICE_API, streaming=False, @@ -1144,6 +1302,7 @@ class TestAppGeneratorExtraCases: app_model=app, workflow=workflow, user=user, + execution_identity=WorkflowExecutionIdentity(user_id=user.id, user_from=UserFrom.END_USER), args={"inputs": {}}, invoke_from=InvokeFrom.SERVICE_API, streaming=False, @@ -1163,6 +1322,7 @@ class TestAppGeneratorExtraCases: workflow_id="workflow-id", ) application_generate_entity = WorkflowAppGenerateEntity.model_construct( + user_from=UserFrom.END_USER, task_id="task", app_config=app_config, inputs={}, @@ -1226,6 +1386,7 @@ class TestAppGeneratorExtraCases: ) existing_trace_manager = SimpleNamespace(app_id="existing-app", user_id="existing-user") application_generate_entity = WorkflowAppGenerateEntity.model_construct( + user_from=UserFrom.END_USER, task_id="task", app_config=app_config, inputs={}, @@ -1298,6 +1459,7 @@ class TestAppGeneratorExtraCases: workflow_id=WORKFLOW_ID, ) application_generate_entity = WorkflowAppGenerateEntity.model_construct( + user_from=UserFrom.END_USER, task_id="task", app_config=app_config, inputs={}, diff --git a/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline.py b/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline.py index ede492abc76..6be87f0ebab 100644 --- a/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline.py +++ b/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline.py @@ -12,7 +12,7 @@ from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import AppAdditionalFeatures, WorkflowUIBasedAppConfig from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.entities.queue_entities import ( QueueAgentLogEvent, QueueErrorEvent, @@ -77,6 +77,7 @@ def _build_workflow_app_config() -> WorkflowUIBasedAppConfig: def _build_generate_entity(run_id: str) -> WorkflowAppGenerateEntity: return WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task-id", app_config=_build_workflow_app_config(), inputs={}, diff --git a/api/tests/unit_tests/core/app/entities/test_app_invoke_entities.py b/api/tests/unit_tests/core/app/entities/test_app_invoke_entities.py index 87c6dac64d5..17be35ee40b 100644 --- a/api/tests/unit_tests/core/app/entities/test_app_invoke_entities.py +++ b/api/tests/unit_tests/core/app/entities/test_app_invoke_entities.py @@ -10,6 +10,7 @@ from core.app.entities.app_invoke_entities import ( CreditUsageCreatedBy, DifyRunContext, InvokeFrom, + UserFrom, WorkflowAppGenerateEntity, get_credit_usage_app_type, get_credit_usage_created_by, @@ -43,6 +44,7 @@ def _build_workflow_app_config(app_mode: AppMode) -> WorkflowUIBasedAppConfig: def _create_workflow_generate_entity(trace_manager: TraceQueueManager | None = None) -> WorkflowAppGenerateEntity: return WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="workflow-task", app_config=_build_workflow_app_config(AppMode.WORKFLOW), inputs={"topic": "serialization"}, diff --git a/api/tests/unit_tests/core/app/entities/test_rag_pipeline_invoke_entities.py b/api/tests/unit_tests/core/app/entities/test_rag_pipeline_invoke_entities.py index dfe39a2acd6..13ae136deb2 100644 --- a/api/tests/unit_tests/core/app/entities/test_rag_pipeline_invoke_entities.py +++ b/api/tests/unit_tests/core/app/entities/test_rag_pipeline_invoke_entities.py @@ -1,4 +1,65 @@ +"""Legacy queue payloads retain the historical pipeline runner role.""" + +import pytest +from pydantic import ValidationError + +from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity, UserFrom from core.app.entities.rag_pipeline_invoke_entities import RagPipelineInvokeEntity +from tests.unit_tests.core.app.apps.pipeline.test_pipeline_runner import _build_app_generate_entity + + +def _load_queued_entity(payload: dict[str, object]) -> RagPipelineGenerateEntity: + envelope = RagPipelineInvokeEntity.model_validate( + { + "pipeline_id": "pipeline", + "application_generate_entity": payload, + "user_id": payload["user_id"], + "tenant_id": "tenant", + "workflow_id": "workflow", + "streaming": False, + } + ) + # Both normal and priority queue consumers validate the nested entity here. + return RagPipelineGenerateEntity.model_validate(envelope.application_generate_entity) + + +@pytest.mark.parametrize("invoke_from", list(InvokeFrom)) +def test_legacy_queue_payload_uses_previous_runner_role(invoke_from: InvokeFrom) -> None: + payload = _build_app_generate_entity().model_dump(mode="json") + payload.pop("user_from") + payload["invoke_from"] = invoke_from.value + + restored = _load_queued_entity(payload) + + expected = UserFrom.ACCOUNT if invoke_from in (InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE) else UserFrom.END_USER + assert restored.user_from == expected + assert "user_from" not in payload + + +@pytest.mark.parametrize("user_from", [UserFrom.ACCOUNT, UserFrom.END_USER]) +def test_queue_preserves_explicit_identity(user_from: UserFrom) -> None: + payload = _build_app_generate_entity().model_dump(mode="json") + payload.update(invoke_from=InvokeFrom.SERVICE_API.value, user_from=user_from.value) + + assert _load_queued_entity(payload).user_from == user_from + + +@pytest.mark.parametrize("field", ["single_iteration_run", "single_loop_run"]) +def test_legacy_debugging_queue_payload_preserves_debugger_role(field: str) -> None: + payload = _build_app_generate_entity().model_dump(mode="json") + payload.pop("user_from") + payload[field] = {"node_id": "node", "inputs": {}} + + assert _load_queued_entity(payload).user_from == UserFrom.ACCOUNT + + +@pytest.mark.parametrize("user_from", [None, "invalid"]) +def test_queue_does_not_repair_invalid_explicit_identity(user_from: str | None) -> None: + payload = _build_app_generate_entity().model_dump(mode="json") + payload["user_from"] = user_from + + with pytest.raises(ValidationError, match="user_from"): + _load_queued_entity(payload) class TestRagPipelineInvokeEntity: diff --git a/api/tests/unit_tests/core/app/layers/test_pause_state_persist_layer.py b/api/tests/unit_tests/core/app/layers/test_pause_state_persist_layer.py index b96c7cc600f..c0f6ff19cc1 100644 --- a/api/tests/unit_tests/core/app/layers/test_pause_state_persist_layer.py +++ b/api/tests/unit_tests/core/app/layers/test_pause_state_persist_layer.py @@ -3,9 +3,15 @@ import json import pytest +from pydantic import ValidationError from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from core.app.layers.pause_state_persist_layer import ( WorkflowResumptionContext, _AdvancedChatAppGenerateEntityWrapper, @@ -28,6 +34,7 @@ def _build_workflow_generate_entity_for_roundtrip() -> WorkflowResumptionContext serialized_graph_runtime_state=serialized_state, generate_entity=_WorkflowGenerateEntityWrapper( entity=WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="workflow-task", app_config=app_config, inputs={"input_key": "input_value"}, @@ -105,3 +112,63 @@ def test_resumption_context_defaults_elapsed_time_for_legacy_snapshots() -> None assert restored.elapsed_time == 0.0 assert restored.get_generate_entity().workflow_execution_timing.elapsed_time == 0.0 + + +@pytest.mark.parametrize("user_from", [UserFrom.ACCOUNT, UserFrom.END_USER]) +def test_resumption_context_preserves_explicit_workflow_user_from(user_from: UserFrom) -> None: + context = _build_workflow_generate_entity_for_roundtrip() + entity = context.get_generate_entity() + assert isinstance(entity, WorkflowAppGenerateEntity) + entity.invoke_from = InvokeFrom.SERVICE_API + entity.user_from = user_from + + restored = WorkflowResumptionContext.loads(context.dumps()).get_generate_entity() + + assert isinstance(restored, WorkflowAppGenerateEntity) + assert restored.user_from == user_from + assert restored.invoke_from == InvokeFrom.SERVICE_API + + +@pytest.mark.parametrize("invoke_from", list(InvokeFrom)) +def test_legacy_workflow_snapshot_uses_previous_runner_role(invoke_from: InvokeFrom) -> None: + payload = _build_workflow_generate_entity_for_roundtrip().model_dump(mode="json") + entity = payload["generate_entity"]["entity"] + entity.pop("user_from") + entity["invoke_from"] = invoke_from.value + + restored = WorkflowResumptionContext.loads(json.dumps(payload)).get_generate_entity() + + expected = UserFrom.ACCOUNT if invoke_from in (InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE) else UserFrom.END_USER + assert restored.user_from == expected + # Normalization must not rewrite the source payload. + assert "user_from" not in entity + + +@pytest.mark.parametrize("field", ["single_iteration_run", "single_loop_run"]) +def test_legacy_single_node_snapshot_preserves_debugger_role(field: str) -> None: + payload = _build_workflow_generate_entity_for_roundtrip().model_dump(mode="json") + entity = payload["generate_entity"]["entity"] + entity.pop("user_from") + entity["invoke_from"] = InvokeFrom.SERVICE_API.value + entity[field] = {"node_id": "node", "inputs": {}} + + restored = WorkflowResumptionContext.loads(json.dumps(payload)).get_generate_entity() + + assert restored.user_from == UserFrom.ACCOUNT + + +@pytest.mark.parametrize("role", [None, "invalid-role"]) +def test_snapshot_does_not_replace_invalid_explicit_identity(role: str | None) -> None: + payload = _build_workflow_generate_entity_for_roundtrip().model_dump(mode="json") + payload["generate_entity"]["entity"]["user_from"] = role + + with pytest.raises(ValidationError): + WorkflowResumptionContext.loads(json.dumps(payload)) + + +def test_new_execution_requires_explicit_identity() -> None: + payload = _build_workflow_generate_entity_for_roundtrip().get_generate_entity().model_dump(mode="json") + payload.pop("user_from") + + with pytest.raises(ValidationError, match="user_from"): + WorkflowAppGenerateEntity.model_validate(payload) diff --git a/api/tests/unit_tests/core/workflow/test_workflow_tool_container_handler.py b/api/tests/unit_tests/core/workflow/test_workflow_tool_container_handler.py index cd5695f2167..84280214813 100644 --- a/api/tests/unit_tests/core/workflow/test_workflow_tool_container_handler.py +++ b/api/tests/unit_tests/core/workflow/test_workflow_tool_container_handler.py @@ -1717,6 +1717,7 @@ class TestWorkflowToolEventDeliveryCases: user.id = "user" layer = WorkflowPersistenceLayer( application_generate_entity=WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task", app_config=WorkflowAppConfig( app_id="outer-app", diff --git a/api/tests/unit_tests/services/test_app_task_service.py b/api/tests/unit_tests/services/test_app_task_service.py index ef347d47861..dcb16e19f8c 100644 --- a/api/tests/unit_tests/services/test_app_task_service.py +++ b/api/tests/unit_tests/services/test_app_task_service.py @@ -20,7 +20,12 @@ import core.app.apps.base_app_queue_manager as queue_module import core.app.apps.execution_coordinator as coordinator_module import services.app_task_service as task_module from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from core.app.layers.pause_state_persist_layer import ( WorkflowResumptionContext, _AdvancedChatAppGenerateEntityWrapper, @@ -680,6 +685,7 @@ def _seed_pause( if app_mode == AppMode.WORKFLOW: wrapper = _WorkflowGenerateEntityWrapper( entity=WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id=str(uuid4()), app_config=config, inputs={}, diff --git a/api/tests/unit_tests/services/test_human_input_service.py b/api/tests/unit_tests/services/test_human_input_service.py index 1489c106d01..b245f91bc86 100644 --- a/api/tests/unit_tests/services/test_human_input_service.py +++ b/api/tests/unit_tests/services/test_human_input_service.py @@ -14,7 +14,7 @@ from sqlalchemy.orm import Session, sessionmaker import services.human_input_service as human_input_service_module from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext, _WorkflowGenerateEntityWrapper from core.repositories.human_input_repository import ( HumanInputFormRecord, @@ -277,6 +277,7 @@ def _build_resumption_context_state(*, options: list[str], workflow_run_id: str) workflow_id="workflow-id", ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task-id", app_config=app_config, inputs={}, diff --git a/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py b/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py index bd7afbdaea5..bc154b0d219 100644 --- a/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py +++ b/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py @@ -14,7 +14,12 @@ import pytest from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from core.app.entities.task_entities import StreamEvent from core.app.layers.pause_state_persist_layer import ( WorkflowResumptionContext, @@ -137,6 +142,7 @@ def _build_resumption_context(task_id: str, *, select_options: list[str] | None workflow_id="workflow-1", ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=task_id, app_config=app_config, inputs={}, @@ -246,6 +252,7 @@ def _build_resumption_context_additional(task_id: str) -> WorkflowResumptionCont workflow_id="workflow-1", ) generate_entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=task_id, app_config=app_config, inputs={}, diff --git a/api/tests/unit_tests/services/test_workflow_execution_identity.py b/api/tests/unit_tests/services/test_workflow_execution_identity.py new file mode 100644 index 00000000000..681f5408302 --- /dev/null +++ b/api/tests/unit_tests/services/test_workflow_execution_identity.py @@ -0,0 +1,30 @@ +"""Caller adapters distinguish actual roles independently of invocation source.""" + +from dataclasses import FrozenInstanceError + +import pytest + +from core.app.entities.app_invoke_entities import UserFrom +from models.model import EndUser +from services.workflow_execution_identity import build_workflow_execution_identity +from tests.helpers.unit.model_factories import make_account + + +def test_account_identity_is_detached_from_orm_state() -> None: + account = make_account(account_id="account-id") + identity = build_workflow_execution_identity(account) + account.id = "changed-id" + + assert identity.user_id == "account-id" + assert identity.user_from == UserFrom.ACCOUNT + with pytest.raises(FrozenInstanceError): + identity.user_id = "changed-id" # type: ignore[read-only] + + +def test_end_user_identity_uses_record_id_not_session_id() -> None: + user = EndUser(id="end-user-id", session_id="external-session") + + identity = build_workflow_execution_identity(user) + + assert identity.user_id == "end-user-id" + assert identity.user_from == UserFrom.END_USER diff --git a/api/tests/unit_tests/services/test_workflow_run_agg.py b/api/tests/unit_tests/services/test_workflow_run_agg.py index c65f97eb74f..fb552e0b601 100644 --- a/api/tests/unit_tests/services/test_workflow_run_agg.py +++ b/api/tests/unit_tests/services/test_workflow_run_agg.py @@ -12,7 +12,7 @@ from core.app.apps.pipeline.pipeline_runner import PipelineRunner from core.app.apps.workflow.app_config_manager import WorkflowAppConfig from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager from core.app.apps.workflow.app_runner import WorkflowAppRunner -from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.entities.queue_entities import ( QueueNodeStartedEvent, QueueWorkflowFailedEvent, @@ -321,6 +321,7 @@ def make_workflow_runner( created_by=user_id, ) entity = WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id=str(uuid4()), app_config=WorkflowAppConfig( tenant_id=tenant_id, app_id=app_id, workflow_id=workflow_id, app_mode=AppMode.WORKFLOW diff --git a/api/tests/unit_tests/services/test_workflow_run_index.py b/api/tests/unit_tests/services/test_workflow_run_index.py index 516b0b7c35a..dffb8e5fe6e 100644 --- a/api/tests/unit_tests/services/test_workflow_run_index.py +++ b/api/tests/unit_tests/services/test_workflow_run_index.py @@ -5,7 +5,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig -from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository from core.workflow.node_factory import DifyNodeFactory @@ -90,6 +90,7 @@ def test_engine_indices_are_available_before_delivery_and_match_persistence( engine = _engine() layer = WorkflowPersistenceLayer( application_generate_entity=WorkflowAppGenerateEntity( + user_from=UserFrom.ACCOUNT, task_id="task", app_config=WorkflowUIBasedAppConfig( tenant_id="tenant", app_id="app", app_mode=AppMode.WORKFLOW, workflow_id="workflow" diff --git a/api/tests/unit_tests/tasks/app_generate/test_workflow_execute_task.py b/api/tests/unit_tests/tasks/app_generate/test_workflow_execute_task.py index c83c492a4b0..8532cb84f5c 100644 --- a/api/tests/unit_tests/tasks/app_generate/test_workflow_execute_task.py +++ b/api/tests/unit_tests/tasks/app_generate/test_workflow_execute_task.py @@ -17,7 +17,12 @@ from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig from core.app.apps.workflow.app_generator import WorkflowAppGenerator -from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity +from core.app.entities.app_invoke_entities import ( + AdvancedChatAppGenerateEntity, + InvokeFrom, + UserFrom, + WorkflowAppGenerateEntity, +) from core.app.layers.pause_state_persist_layer import ( WorkflowResumptionContext, _AdvancedChatAppGenerateEntityWrapper, @@ -69,6 +74,7 @@ def _build_advanced_chat_generate_entity(conversation_id: str | None) -> Advance def _build_workflow_generate_entity(stream: bool) -> WorkflowAppGenerateEntity: return WorkflowAppGenerateEntity( + user_from=UserFrom.END_USER, task_id="task-id", inputs={}, files=[],