From a4d64bcdddf1f080ff61e53029c3f64474d48cf6 Mon Sep 17 00:00:00 2001 From: QuantumGhost Date: Tue, 10 Nov 2026 09:00:41 +0000 Subject: [PATCH] feat(hitl): improve forms, notifications, and IM contact sync (#45945) Co-authored-by: zxhlyh --- .../common/human_input_v2_contracts.py | 8 +- .../console/human_input_v2/contact.py | 17 +- .../app/apps/advanced_chat/app_generator.py | 15 +- api/core/app/apps/base_app_generator.py | 4 + api/core/app/apps/workflow/app_generator.py | 16 +- .../human_input_v2/im_integration/ports.py | 6 +- .../im_integration/sync_records.py | 34 +- ...00-6b7e2a9c4d10_add_im_sync_sys_user_id.py | 42 +++ api/models/human_input_v2.py | 13 +- api/openapi/markdown/console-openapi.md | 10 +- .../human_input_v2/form_repository.py | 8 + .../human_input_v2/im_integration/mappers.py | 39 ++- .../im_integration/repository.py | 12 +- .../sqlalchemy_form_repository.py | 12 + .../im_contact_sync/composition.py | 7 +- .../im_contact_sync/coordinator.py | 3 +- .../human_input_v2/im_contact_sync/service.py | 50 ++- .../human_input_v2/submission_service.py | 39 ++- .../app_generate/workflow_execute_task.py | 11 + api/tasks/async_workflow_tasks.py | 7 + .../workspace/test_im_contact_sync_api.py | 3 +- .../test_im_control_plane_concurrency.py | 3 +- .../human_input_v2/test_stale_im_sync_runs.py | 37 +- .../common/test_human_input_v2_contracts.py | 1 + .../workspace/test_im_contact_sync_api.py | 203 ++++++----- .../apps/advanced_chat/test_app_generator.py | 67 +++- .../app/apps/workflow/test_app_generator.py | 145 ++++++++ .../test_im_sync_reconciliation.py | 3 +- .../im_integration/test_im_sync_queries.py | 3 +- .../im_integration/test_mappers.py | 31 +- .../im_integration/test_repository.py | 33 +- .../test_im_channel_reconciliation_service.py | 25 +- .../test_im_contact_sync_composition.py | 3 +- .../test_im_contact_sync_worker.py | 3 +- .../human_input_v2/test_im_sync_service.py | 112 +++++- .../human_input_v2/test_submission_service.py | 123 ++++++- oxlint-suppressions.json | 10 - .../api/console/workspaces/orpc.gen.ts | 8 +- .../api/console/workspaces/types.gen.ts | 2 + .../api/console/workspaces/zod.gen.ts | 6 +- .../components/base/prompt-editor/index.tsx | 9 + .../__tests__/component-ui.spec.tsx | 113 +++--- .../__tests__/input-field.spec.tsx | 84 ++++- .../plugins/hitl-input-block/component-ui.tsx | 10 +- .../plugins/hitl-input-block/input-field.tsx | 72 +++- .../prompt-editor/prompt-editor-content.tsx | 9 + .../__tests__/index.hitl-v2.browser.spec.tsx | 11 +- .../message-template-subject.browser.spec.tsx | 125 +++++++ .../__tests__/message-template.spec.tsx | 196 ++++++++--- .../human-input-v2/__tests__/panel.spec.tsx | 24 +- .../recipients-keyboard.browser.spec.tsx | 1 + .../__tests__/recipients.spec.tsx | 58 ++++ .../components/message-template-subject.tsx | 228 +++++++++++++ .../components/message-template.tsx | 15 +- .../human-input-v2/components/recipients.tsx | 58 +++- .../workflow/nodes/human-input-v2/panel.tsx | 4 + .../form-content-preview.hitl-v2.spec.tsx | 322 +++++++++++++++++- .../__tests__/form-content.browser.spec.tsx | 59 +++- .../components/form-content-preview-inputs.ts | 149 ++++++++ .../components/form-content-preview.tsx | 143 +++++++- .../human-input/shared/panel-sections.tsx | 2 + .../im-platform/__tests__/fixtures/http.ts | 1 + .../__tests__/management-surface.spec.tsx | 24 +- .../nested-dialog-backdrops.browser.spec.tsx | 2 + .../im-platform/management-surface.tsx | 23 +- web/i18n/locales/ar-TN/workflow.json | 1 + web/i18n/locales/az-AZ/workflow.json | 1 + web/i18n/locales/de-DE/workflow.json | 1 + web/i18n/locales/en-US/workflow.json | 1 + web/i18n/locales/es-ES/workflow.json | 1 + web/i18n/locales/fa-IR/workflow.json | 1 + web/i18n/locales/fr-FR/workflow.json | 1 + web/i18n/locales/hi-IN/workflow.json | 1 + web/i18n/locales/id-ID/workflow.json | 1 + web/i18n/locales/it-IT/workflow.json | 1 + web/i18n/locales/ja-JP/workflow.json | 1 + web/i18n/locales/ko-KR/workflow.json | 1 + web/i18n/locales/lo-LA/workflow.json | 1 + web/i18n/locales/nl-NL/workflow.json | 1 + web/i18n/locales/pl-PL/workflow.json | 1 + web/i18n/locales/pt-BR/workflow.json | 1 + web/i18n/locales/ro-RO/workflow.json | 1 + web/i18n/locales/ru-RU/workflow.json | 1 + web/i18n/locales/sl-SI/workflow.json | 1 + web/i18n/locales/th-TH/workflow.json | 1 + web/i18n/locales/tr-TR/workflow.json | 1 + web/i18n/locales/uk-UA/workflow.json | 1 + web/i18n/locales/vi-VN/workflow.json | 1 + web/i18n/locales/zh-Hans/workflow.json | 1 + web/i18n/locales/zh-Hant/workflow.json | 1 + web/i18n/resources.generated.d.ts | 1 + 91 files changed, 2590 insertions(+), 352 deletions(-) create mode 100644 api/migrations/versions/2026_11_09_1200-6b7e2a9c4d10_add_im_sync_sys_user_id.py create mode 100644 web/app/components/workflow/nodes/human-input-v2/__tests__/message-template-subject.browser.spec.tsx create mode 100644 web/app/components/workflow/nodes/human-input-v2/components/message-template-subject.tsx create mode 100644 web/app/components/workflow/nodes/human-input/components/form-content-preview-inputs.ts diff --git a/api/controllers/common/human_input_v2_contracts.py b/api/controllers/common/human_input_v2_contracts.py index b3cbf70037a..562a7460d6b 100644 --- a/api/controllers/common/human_input_v2_contracts.py +++ b/api/controllers/common/human_input_v2_contracts.py @@ -134,6 +134,7 @@ class HumanInputContactSummary(BaseModel): id: ContactId = Field(description="Unique contact identifier.") name: str = Field(description="Display name shown in the contact directory.") + email: str | None = Field(default=None, description="Primary contact email if one exists.") avatar_url: str = Field(default="", description="URL of the contact's avatar.") created_at: Timestamp = Field(description="Timestamp when the contact was created.") @@ -279,12 +280,15 @@ class IMSyncRunResultCounts(ResponseModel): class IMSyncRun(ResponseModel): """One IM sync run snapshot. - The latest-only UI displays ``finished_at`` as the explicit sync time. The - transport contract intentionally does not expose a ``started_by`` actor. + The latest-only UI displays ``finished_at`` as the explicit sync time. + The initiator display name is resolved from the initiating Dify Account. """ id: IMSyncRunId = Field(description="Unique sync run identifier.") status: IMSyncRunStatus = Field(description="Current lifecycle state of the sync run.") + initiator_display_name: str = Field( + description="Current initiating Account name; empty for EE administrators or an unavailable Account." + ) started_at: Timestamp | None = Field( default=None, description="Unix timestamp in seconds when the sync run started." ) diff --git a/api/controllers/console/human_input_v2/contact.py b/api/controllers/console/human_input_v2/contact.py index db083c55fe8..1e6adcac0c7 100644 --- a/api/controllers/console/human_input_v2/contact.py +++ b/api/controllers/console/human_input_v2/contact.py @@ -76,10 +76,10 @@ from core.human_input_v2.im_integration import ( ContactIMBindingView, IMBindingCommandError, IMBindingCommandErrorCode, - IMSyncRun, SyncContactSnapshot, SyncResultFact, ) +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator from core.human_input_v2.shared import ( AccountId, ContactId, @@ -123,6 +123,7 @@ from services.human_input_v2.im_contact_sync.service import ( IMSyncDispatchUnavailableError, IMSyncRevisionChangedError, IMSyncRunNotFoundError, + IMSyncRunSummary, ) from services.human_input_v2.node_data_migration import MigrationNode, NodeDataMigrationFailure @@ -237,10 +238,13 @@ def _im_application_error_response(error: _IMApplicationError) -> tuple[dict[str ) -def _sync_run_payload(run: IMSyncRun) -> dict[str, object]: +def _sync_run_payload(summary: IMSyncRunSummary) -> dict[str, object]: + """Serialize a persisted run with the currently resolved Account name.""" + run = summary.run return { "id": run.id, "status": run.status, + "initiator_display_name": summary.initiator_display_name, "started_at": run.started_at, "finished_at": run.finished_at, "error_message": run.error_message, @@ -272,11 +276,13 @@ def _avatar_url(avatar_file_id: str | None) -> str: def _sync_contact_payload(contact: SyncContactSnapshot, fallback_created_at) -> dict[str, object]: + """Serialize the frozen Contact profile without consulting current state.""" # Historical result snapshots predate the Contact creation-time field. created_at = contact.created_at if contact.created_at is not None else fallback_created_at return { "id": contact.contact_id, "name": contact.name, + "email": contact.email, "avatar_url": _avatar_url(contact.avatar_file_id), "created_at": created_at, } @@ -369,10 +375,12 @@ def _contact_option_payload(view: ContactWithIMBindings) -> dict[str, object]: def _contact_summary_payload(view: ContactWithIMBindings) -> dict[str, object]: + """Project one resolved Contact, including its available primary email.""" contact = view.contact return { "id": contact.id, "name": contact.name, + "email": contact.email, "avatar_url": _contact_avatar_url(contact), "created_at": contact.created_at, } @@ -655,10 +663,11 @@ class WorkspaceIMSyncRunsApi(Resource): @with_current_user @with_current_tenant_id def post(self, tenant_id: str, current_user: Account): + """Start or reuse a workspace run using the authenticated Account ID.""" try: run = build_im_contact_sync_application().sync_service.create_or_get_active_run( _workspace_scope(tenant_id), - AccountId(current_user.id), + AccountIMSyncInitiator(AccountId(current_user.id)), ) except ( IMChannelNotConfiguredError, @@ -675,7 +684,7 @@ class WorkspaceLatestIMSyncRunApi(Resource): @console_ns.doc( description=( "Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; " - "the response does not include started_by." + "initiator_display_name is resolved from the initiating Account." ) ) @console_ns.response(200, "Success", console_ns.models[GetLatestIMSyncRunResponse.__name__]) diff --git a/api/core/app/apps/advanced_chat/app_generator.py b/api/core/app/apps/advanced_chat/app_generator.py index ba669dc77ff..17baef71fe0 100644 --- a/api/core/app/apps/advanced_chat/app_generator.py +++ b/api/core/app/apps/advanced_chat/app_generator.py @@ -4,7 +4,7 @@ import contextvars import logging import threading import uuid -from collections.abc import Generator, Mapping +from collections.abc import Callable, Generator, Mapping from typing import TYPE_CHECKING, Any, Literal, overload from flask import Flask, current_app @@ -29,6 +29,7 @@ from core.app.apps.advanced_chat.generate_task_pipeline import ( MessageSnapshot, WorkflowSnapshot, ) +from core.app.apps.base_app_generator import noop_restore_file_access from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom from core.app.apps.draft_variable_saver import DraftVariableSaverFactory from core.app.apps.exc import GenerateTaskStoppedError @@ -291,10 +292,15 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator): graph_runtime_state: RuntimeState, pause_state_config: PauseStateLayerConfig | None = None, response_stream_filter: ResponseStreamFilter | None = None, + restore_file_access: Callable[[], None] = noop_restore_file_access, ): """ Resume a paused advanced chat execution. + ``restore_file_access`` restores host-owned grants inside the new + execution scope before worker contexts are copied. The default callback + leaves existing access rules unchanged. + ``trace_scope`` is transient and excluded from generate-entity serialization, so resumed executions create it again before persistence layers receive the entity. """ @@ -324,6 +330,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator): pause_state_config=pause_state_config, graph_runtime_state=graph_runtime_state, response_stream_filter=response_stream_filter, + restore_file_access=restore_file_access, session=session, ) @@ -532,10 +539,14 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator): pause_state_config: PauseStateLayerConfig | None = None, graph_runtime_state: RuntimeState | None = None, response_stream_filter: ResponseStreamFilter | None = None, + restore_file_access: Callable[[], None] = noop_restore_file_access, ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any]]: """ Generate App response. + On resume, HITL submissions must already be committed. Restore their + EndUser file grants before capturing worker and response contexts. + :param workflow: Workflow :param user: account or end user :param invoke_from: invoke from source @@ -552,6 +563,8 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator): invoke_from=invoke_from, ): grant_file_default_access(application_generate_entity.app_config.variables) + # Restore grants before copying contexts for both workers and responses. + restore_file_access() is_first_conversation = conversation is None if conversation is not None and message is not None: diff --git a/api/core/app/apps/base_app_generator.py b/api/core/app/apps/base_app_generator.py index 4a23bb71af4..9ff2547fdf0 100644 --- a/api/core/app/apps/base_app_generator.py +++ b/api/core/app/apps/base_app_generator.py @@ -36,6 +36,10 @@ logger = logging.getLogger(__name__) _WORKER_THREAD_JOIN_TIMEOUT_SECONDS = 300 +def noop_restore_file_access() -> None: + """Leave the current file-access scope unchanged when no restoration is needed.""" + + @final class _DebuggerDraftVariableSaver: """Adapter that binds SQLAlchemy session setup outside the saver port.""" diff --git a/api/core/app/apps/workflow/app_generator.py b/api/core/app/apps/workflow/app_generator.py index e5223b9de35..7c72d236c90 100644 --- a/api/core/app/apps/workflow/app_generator.py +++ b/api/core/app/apps/workflow/app_generator.py @@ -4,7 +4,7 @@ import contextvars import logging import threading import uuid -from collections.abc import Generator, Mapping, Sequence +from collections.abc import Callable, Generator, Mapping, Sequence from typing import TYPE_CHECKING, Any, Literal, overload from flask import Flask, current_app @@ -15,7 +15,7 @@ from sqlalchemy.orm import Session, sessionmaker import contexts from configs import dify_config from core.app.app_config.features.file_upload.manager import FileUploadConfigManager -from core.app.apps.base_app_generator import BaseAppGenerator +from core.app.apps.base_app_generator import BaseAppGenerator, noop_restore_file_access from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom from core.app.apps.draft_variable_saver import DraftVariableSaverFactory from core.app.apps.exc import GenerateTaskStoppedError @@ -300,10 +300,15 @@ class WorkflowAppGenerator(BaseAppGenerator): pause_state_config: PauseStateLayerConfig | None = None, variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER, response_stream_filter: ResponseStreamFilter | None = None, + restore_file_access: Callable[[], None] = noop_restore_file_access, ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any]]: """ Resume a paused workflow execution using the persisted runtime state. + ``restore_file_access`` restores host-owned grants inside the new + execution scope before worker contexts are copied. The default callback + leaves existing access rules unchanged. + ``trace_scope`` is transient and excluded from generate-entity serialization, so resumed executions create it again before persistence layers receive the entity. """ @@ -334,6 +339,7 @@ class WorkflowAppGenerator(BaseAppGenerator): graph_runtime_state=graph_runtime_state, pause_state_config=pause_state_config, response_stream_filter=response_stream_filter, + restore_file_access=restore_file_access, ) def _generate( @@ -353,10 +359,14 @@ class WorkflowAppGenerator(BaseAppGenerator): graph_runtime_state: RuntimeState | None = None, pause_state_config: PauseStateLayerConfig | None = None, response_stream_filter: ResponseStreamFilter | None = None, + restore_file_access: Callable[[], None] = noop_restore_file_access, ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any]]: """ Generate App response. + On resume, HITL submissions must already be committed. Restore their + EndUser file grants before capturing worker and response contexts. + :param app_model: App :param workflow: Workflow :param user: account or end user @@ -372,6 +382,8 @@ class WorkflowAppGenerator(BaseAppGenerator): invoke_from=invoke_from, ): grant_file_default_access(application_generate_entity.app_config.variables) + # Restore grants before copying contexts for both workers and responses. + restore_file_access() # init queue manager queue_manager = WorkflowAppQueueManager( task_id=application_generate_entity.task_id, diff --git a/api/core/human_input_v2/im_integration/ports.py b/api/core/human_input_v2/im_integration/ports.py index e0dba15c5b9..51b77631ee8 100644 --- a/api/core/human_input_v2/im_integration/ports.py +++ b/api/core/human_input_v2/im_integration/ports.py @@ -13,7 +13,6 @@ from pydantic import NaiveDatetime from core.human_input_v2.entities import IMProvider, IMSyncResultType from core.human_input_v2.shared import ( - AccountId, ContactId, DirectoryScope, IMIdentityId, @@ -24,6 +23,7 @@ from repositories.human_input_v2.im_channel_repository import IMChannelId from .sync_reconciliation import ReconciliationReasonCode from .sync_records import ( IMChannelRevision, + IMSyncInitiator, IMSyncRun, StaleRevision, SynchronizedIMIdentityPage, @@ -103,10 +103,10 @@ class IMSyncRepository(Protocol): *, organization_scope: DirectoryScope, sync_run_id: IMSyncRunId, - started_by_account_id: AccountId | None, + started_by: IMSyncInitiator, now: NaiveDatetime, ) -> ActiveRunDecision: - """Lock Channel and return at most one active run.""" + """Lock Channel and create or reuse a run, preserving its original initiator.""" ... def load_sync_run(self, sync_run_id: IMSyncRunId) -> IMSyncRun | None: diff --git a/api/core/human_input_v2/im_integration/sync_records.py b/api/core/human_input_v2/im_integration/sync_records.py index 835627af2ca..0733490cb9a 100644 --- a/api/core/human_input_v2/im_integration/sync_records.py +++ b/api/core/human_input_v2/im_integration/sync_records.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass, replace +from typing import NewType from pydantic import NaiveDatetime @@ -127,9 +128,33 @@ class SynchronizedIMIdentityPage: total: int +SysUserId = NewType("SysUserId", str) + + +@dataclass(frozen=True, slots=True) +class AccountIMSyncInitiator: + """A Dify Account that initiated a CE/SaaS synchronization.""" + + account_id: AccountId + + +@dataclass(frozen=True, slots=True) +class SysUserIMSyncInitiator: + """An EE administrator identified by the upstream sys_users.id UUID.""" + + sys_user_id: SysUserId + + +type IMSyncInitiator = AccountIMSyncInitiator | SysUserIMSyncInitiator + + @dataclass(frozen=True, slots=True) class IMSyncRun: - """Independent sync aggregate that captures one complete Channel revision.""" + """Capture one Channel revision and either an Account or EE administrator. + + Every run has exactly one initiating identity, represented by one of the + two variants. Missing initiators are not valid synchronization state. + """ id: IMSyncRunId channel_revision: IMChannelRevision @@ -140,7 +165,7 @@ class IMSyncRun: failed_count: int removed_count: int skipped_count: int - started_by_account_id: AccountId | None + started_by: IMSyncInitiator started_at: NaiveDatetime | None finished_at: NaiveDatetime | None error_code: str | None @@ -155,9 +180,10 @@ class IMSyncRun: sync_run_id: IMSyncRunId, channel_revision: IMChannelRevision, provider: IMProvider, - started_by_account_id: AccountId | None, + started_by: IMSyncInitiator, now: NaiveDatetime, ) -> IMSyncRun: + """Create a queued run from one trusted initiating identity, without I/O.""" return cls( id=sync_run_id, channel_revision=channel_revision, @@ -168,7 +194,7 @@ class IMSyncRun: failed_count=0, removed_count=0, skipped_count=0, - started_by_account_id=started_by_account_id, + started_by=started_by, started_at=None, finished_at=None, error_code=None, diff --git a/api/migrations/versions/2026_11_09_1200-6b7e2a9c4d10_add_im_sync_sys_user_id.py b/api/migrations/versions/2026_11_09_1200-6b7e2a9c4d10_add_im_sync_sys_user_id.py new file mode 100644 index 00000000000..5db4cf5a3c1 --- /dev/null +++ b/api/migrations/versions/2026_11_09_1200-6b7e2a9c4d10_add_im_sync_sys_user_id.py @@ -0,0 +1,42 @@ +"""Add the upstream EE synchronization initiator identity. + +Revision ID: 6b7e2a9c4d10 +Revises: 53c10126d551 +Create Date: 2026-11-09 12:00:00 + +""" + +import sqlalchemy as sa +from alembic import op + +from models.types import StringUUID + +revision = "6b7e2a9c4d10" +down_revision = "53c10126d551" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """Add the EE UUID and require exactly one initiating identity per run.""" + with op.batch_alter_table("human_input_im_sync_runs") as batch_op: + batch_op.add_column( + sa.Column( + "started_by_sys_user_id", + StringUUID, + nullable=True, + comment="Logical reference to upstream EE sys_users.id; mutually exclusive with started_by_account_id.", + ) + ) + batch_op.create_check_constraint( + "initiator_exclusive", + "(started_by_account_id IS NOT NULL AND started_by_sys_user_id IS NULL) OR " + "(started_by_account_id IS NULL AND started_by_sys_user_id IS NOT NULL)", + ) + + +def downgrade() -> None: + """Remove the EE initiator reference while preserving existing Account IDs.""" + with op.batch_alter_table("human_input_im_sync_runs") as batch_op: + batch_op.drop_constraint("initiator_exclusive", type_="check") + batch_op.drop_column("started_by_sys_user_id") diff --git a/api/models/human_input_v2.py b/api/models/human_input_v2.py index 557f0077979..f267a50ac62 100644 --- a/api/models/human_input_v2.py +++ b/api/models/human_input_v2.py @@ -768,6 +768,11 @@ class HumanInputIMSyncRun(DefaultFieldsDCMixin, TypeBase): __tablename__ = "human_input_im_sync_runs" __table_args__ = ( sa.CheckConstraint("channel_config_version > 0", name="captured_version_positive"), + sa.CheckConstraint( + "(started_by_account_id IS NOT NULL AND started_by_sys_user_id IS NULL) OR " + "(started_by_account_id IS NULL AND started_by_sys_user_id IS NOT NULL)", + name="initiator_exclusive", + ), sa.CheckConstraint( "added_count >= 0 AND not_matched_count >= 0 AND failed_count >= 0 AND removed_count >= 0 " "AND skipped_count >= 0", @@ -836,7 +841,13 @@ class HumanInputIMSyncRun(DefaultFieldsDCMixin, TypeBase): StringUUID, nullable=True, default=None, - comment="Logical foreign key to accounts.id for the administrator who started this run.", + comment="Logical accounts.id reference for CE/SaaS initiators; null only for EE sys_user initiators.", + ) + started_by_sys_user_id: Mapped[str | None] = mapped_column( + StringUUID, + nullable=True, + default=None, + comment="Logical reference to upstream EE sys_users.id; mutually exclusive with started_by_account_id.", ) started_at: Mapped[datetime | None] = mapped_column( sa.DateTime, diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md index b8f3d495e6e..2c8b2f6032a 100644 --- a/api/openapi/markdown/console-openapi.md +++ b/api/openapi/markdown/console-openapi.md @@ -15220,6 +15220,8 @@ Set or reset the IM override for a contact. This endpoint is used to override th | default | Error | **application/json**: [ErrorResponse](#errorresponse)
| ### [POST] /workspaces/current/human-input/v2/im-sync-runs +**Start or reuse a workspace run using the authenticated Account ID** + #### Responses | Code | Description | Schema | @@ -15231,7 +15233,7 @@ Set or reset the IM override for a contact. This endpoint is used to override th | default | Error | **application/json**: [ErrorResponse](#errorresponse)
| ### [GET] /workspaces/current/human-input/v2/im-sync-runs/latest -Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; the response does not include started_by. +Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; initiator_display_name is resolved from the initiating Account. #### Responses @@ -26777,6 +26779,7 @@ A trimmed version of `HumanInputContact` that only includes the fields needed fo | ---- | ---- | ----------- | -------- | | avatar_url | string | URL of the contact's avatar. | No | | created_at | integer | Timestamp when the contact was created. | Yes | +| email | string | Primary contact email if one exists. | No | | id | string | Unique contact identifier. | Yes | | name | string | Display name shown in the contact directory. | Yes | @@ -27139,8 +27142,8 @@ Stable reconciliation bucket for one synchronized directory entry. One IM sync run snapshot. -The latest-only UI displays ``finished_at`` as the explicit sync time. The -transport contract intentionally does not expose a ``started_by`` actor. +The latest-only UI displays ``finished_at`` as the explicit sync time. +The initiator display name is resolved from the initiating Dify Account. | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | @@ -27149,6 +27152,7 @@ transport contract intentionally does not expose a ``started_by`` actor. | error_message | string | Terminal error message. Present only when the sync run status is `failed`. | No | | finished_at | integer | Unix timestamp in seconds when the sync run finished. This is the sync time displayed by the latest-only UI and is None while the run is unfinished. | No | | id | string | Unique sync run identifier. | Yes | +| initiator_display_name | string | Current initiating Account name; empty for EE administrators or an unavailable Account. | Yes | | provider | [IMProvider](#improvider) | IM provider associated with the sync run. | Yes | | result_counts | [IMSyncRunResultCounts](#imsyncrunresultcounts) | Aggregate reconciliation counts for the current run snapshot. | Yes | | started_at | integer | Unix timestamp in seconds when the sync run started. | No | diff --git a/api/repositories/human_input_v2/form_repository.py b/api/repositories/human_input_v2/form_repository.py index 918bf7b2b1e..2378f1a1b1e 100644 --- a/api/repositories/human_input_v2/form_repository.py +++ b/api/repositories/human_input_v2/form_repository.py @@ -122,6 +122,14 @@ class FormRepository(Protocol): """ ... + def get_submitted_forms(self, *, workflow_run_id: str) -> tuple[Form, ...]: + """Read accepted runtime submissions for one run in the bound tenant and app. + + Exclude waiting, timed-out, expired, and delivery-test forms. Return + detached snapshots without modifying forms or their submissions. + """ + ... + def create_form(self, params: FormCreateParams) -> Form: """Create an unsubmitted form with generated ID and timestamps. diff --git a/api/repositories/human_input_v2/im_integration/mappers.py b/api/repositories/human_input_v2/im_integration/mappers.py index b455d441c3a..2f6dd04f96b 100644 --- a/api/repositories/human_input_v2/im_integration/mappers.py +++ b/api/repositories/human_input_v2/im_integration/mappers.py @@ -1,6 +1,7 @@ """Explicit bidirectional mappings for IM domain values and ORM records.""" from datetime import datetime +from typing import assert_never from pydantic import NaiveDatetime @@ -18,6 +19,12 @@ from core.human_input_v2.im_integration import ( SyncResultFact, ) from core.human_input_v2.im_integration.adapters.entities import ProviderUserId +from core.human_input_v2.im_integration.sync_records import ( + AccountIMSyncInitiator, + IMSyncInitiator, + SysUserId, + SysUserIMSyncInitiator, +) from core.human_input_v2.shared import ( AccountId, ContactId, @@ -48,6 +55,19 @@ def _timestamp(value: datetime) -> NaiveDatetime: return ensure_naive_utc(value) +def _sync_initiator_from_record(record: HumanInputIMSyncRun) -> IMSyncInitiator: + """Restore the identity guaranteed by the database's exactly-one constraint.""" + account_id = record.started_by_account_id + sys_user_id = record.started_by_sys_user_id + if account_id is not None: + if sys_user_id is not None: + raise AssertionError("IM sync run must contain exactly one initiating identity") + return AccountIMSyncInitiator(AccountId(account_id)) + if sys_user_id is None: + raise AssertionError("IM sync run must contain exactly one initiating identity") + return SysUserIMSyncInitiator(SysUserId(sys_user_id)) + + def sync_run_from_record(record: HumanInputIMSyncRun) -> IMSyncRun: """Map one sync run record into its independent aggregate.""" @@ -61,9 +81,7 @@ def sync_run_from_record(record: HumanInputIMSyncRun) -> IMSyncRun: failed_count=record.failed_count, removed_count=record.removed_count, skipped_count=record.skipped_count, - started_by_account_id=( - AccountId(record.started_by_account_id) if record.started_by_account_id is not None else None - ), + started_by=_sync_initiator_from_record(record), started_at=_timestamp(record.started_at) if record.started_at is not None else None, finished_at=_timestamp(record.finished_at) if record.finished_at is not None else None, error_code=record.error_code, @@ -74,7 +92,17 @@ def sync_run_from_record(record: HumanInputIMSyncRun) -> IMSyncRun: def sync_run_to_record(run: IMSyncRun) -> HumanInputIMSyncRun: - """Map one sync aggregate into a detached record.""" + """Map either initiating identity into exactly one column of a detached record.""" + + match run.started_by: + case AccountIMSyncInitiator(account_id): + started_by_account_id = str(account_id) + started_by_sys_user_id = None + case SysUserIMSyncInitiator(sys_user_id): + started_by_account_id = None + started_by_sys_user_id = str(sys_user_id) + case unexpected: + assert_never(unexpected) record = HumanInputIMSyncRun( channel_id=run.channel_revision.channel_id, @@ -86,7 +114,8 @@ def sync_run_to_record(run: IMSyncRun) -> HumanInputIMSyncRun: failed_count=run.failed_count, removed_count=run.removed_count, skipped_count=run.skipped_count, - started_by_account_id=str(run.started_by_account_id) if run.started_by_account_id is not None else None, + started_by_account_id=started_by_account_id, + started_by_sys_user_id=started_by_sys_user_id, started_at=run.started_at if run.started_at is not None else None, finished_at=run.finished_at if run.finished_at is not None else None, error_code=run.error_code, diff --git a/api/repositories/human_input_v2/im_integration/repository.py b/api/repositories/human_input_v2/im_integration/repository.py index 3858659a4a6..6aac4276448 100644 --- a/api/repositories/human_input_v2/im_integration/repository.py +++ b/api/repositories/human_input_v2/im_integration/repository.py @@ -19,7 +19,8 @@ from core.human_input_v2.im_integration import ( SyncResultFact, SyncResultPage, ) -from core.human_input_v2.shared import AccountId, IMSyncResultId, IMSyncRunId +from core.human_input_v2.im_integration.sync_records import IMSyncInitiator +from core.human_input_v2.shared import IMSyncResultId, IMSyncRunId from libs.uuid_utils import uuidv7 from models.human_input_v2 import HumanInputIMChannel, HumanInputIMSyncResult, HumanInputIMSyncRun from repositories.human_input_v2.im_binding_repository import IMBindingRepository @@ -42,9 +43,14 @@ class SQLAlchemyIMControlPlaneRepository: self, *, sync_run_id: IMSyncRunId, - started_by_account_id: AccountId | None, + started_by: IMSyncInitiator, now: NaiveDatetime, ) -> ActiveRunDecision: + """Persist one initiating identity or return the unchanged active run. + + The caller owns the transaction. The Channel lock serializes creation; + a stale revision returns without creating a run. + """ revision = IMChannelRevision( str(self._channel.id), self._channel.config_version, @@ -78,7 +84,7 @@ class SQLAlchemyIMControlPlaneRepository: sync_run_id=sync_run_id, channel_revision=revision, provider=self._channel.provider, - started_by_account_id=started_by_account_id, + started_by=started_by, now=now, ) record = sync_run_to_record(run) diff --git a/api/repositories/human_input_v2/sqlalchemy_form_repository.py b/api/repositories/human_input_v2/sqlalchemy_form_repository.py index 3ff7816670f..c1a37fdb59c 100644 --- a/api/repositories/human_input_v2/sqlalchemy_form_repository.py +++ b/api/repositories/human_input_v2/sqlalchemy_form_repository.py @@ -210,6 +210,18 @@ class SQLAlchemyFormRepository(FormRepository): ) return tuple(_form_from_record(record) for record in records) + @override + def get_submitted_forms(self, *, workflow_run_id: str) -> tuple[Form, ...]: + """Return only accepted runtime forms in the bound owner scope without writes.""" + records = self._session.scalars( + self._query().where( + HumanInputForm.workflow_run_id == workflow_run_id, + HumanInputForm.form_kind == HumanInputFormKind.RUNTIME, + HumanInputForm.status == HumanInputFormStatus.SUBMITTED, + ) + ) + return tuple(_form_from_record(record) for record in records) + @override def create_form(self, params: FormCreateParams) -> Form: execution_ids = (params.workflow_run_id, params.node_execution_id) diff --git a/api/services/human_input_v2/im_contact_sync/composition.py b/api/services/human_input_v2/im_contact_sync/composition.py index 87073b9d423..636723a99c9 100644 --- a/api/services/human_input_v2/im_contact_sync/composition.py +++ b/api/services/human_input_v2/im_contact_sync/composition.py @@ -13,6 +13,7 @@ from core.human_input_v2.im_integration.adapters.factory import build_im_provide from core.human_input_v2.shared import DeploymentScope, DirectoryScope, IMSyncRunId, TenantId, WorkspaceScope from extensions.ext_database import db from extensions.ext_key_provider import key_provider_manager +from repositories.account.repository import SQLAlchemyAccountRepository from repositories.human_input_v2.contact import Contact, ContactQuery, ContactType from repositories.human_input_v2.im_channel_repository import IMChannel from repositories.human_input_v2.sqlalchemy_contact_repository import SQLAlchemyContactRepository @@ -103,7 +104,9 @@ def build_im_contact_sync_application( deployment_adapter_factory: IMChannelAdapterFactory | None = None, ) -> IMContactSyncApplication: sessions = session_maker or sessionmaker(bind=db.engine, expire_on_commit=False) - sync_service = IMSyncService(sessions, _resolve_channel, _dispatch) + sync_service = IMSyncService( + sessions, _resolve_channel, _dispatch, account_query=SQLAlchemyAccountRepository(sessions) + ) worker = IMContactSyncWorker( sync_service, _reconciliation_factory( @@ -137,7 +140,7 @@ def build_im_contact_sync_worker( def build_im_sync_service(*, session_maker: sessionmaker[Session] | None = None) -> IMSyncService: sessions = session_maker or sessionmaker(bind=db.engine, expire_on_commit=False) - return IMSyncService(sessions, _resolve_channel, _dispatch) + return IMSyncService(sessions, _resolve_channel, _dispatch, account_query=SQLAlchemyAccountRepository(sessions)) def _resolve_channel(session: Session, owner_scope: DirectoryScope) -> IMChannel | None: diff --git a/api/services/human_input_v2/im_contact_sync/coordinator.py b/api/services/human_input_v2/im_contact_sync/coordinator.py index c0ae40bae0f..5f4badcd411 100644 --- a/api/services/human_input_v2/im_contact_sync/coordinator.py +++ b/api/services/human_input_v2/im_contact_sync/coordinator.py @@ -39,6 +39,7 @@ from core.human_input_v2.im_integration import ( ) from core.human_input_v2.im_integration.adapters import IMProviderAdapter from core.human_input_v2.im_integration.adapters.entities import DirectoryEntry, DirectoryReadFailure, ProviderUserId +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator from core.human_input_v2.shared import ( AccountId, ContactId, @@ -236,7 +237,7 @@ class IMChannelReconciliationService: # tenant only for Workspace Channels. if self._tenant_id is None or not any(entry.avatar is not None for entry in entries): return {} - created_by = run.started_by_account_id + created_by = run.started_by.account_id if isinstance(run.started_by, AccountIMSyncInitiator) else None if created_by is None: with self._session_factory() as session: owner_id = session.scalar( diff --git a/api/services/human_input_v2/im_contact_sync/service.py b/api/services/human_input_v2/im_contact_sync/service.py index 25207f7423d..3df6183c447 100644 --- a/api/services/human_input_v2/im_contact_sync/service.py +++ b/api/services/human_input_v2/im_contact_sync/service.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging from collections.abc import Callable +from dataclasses import dataclass from pydantic import NaiveDatetime from sqlalchemy.orm import Session, sessionmaker @@ -15,7 +16,8 @@ from core.human_input_v2.im_integration import ( SynchronizedIMIdentityPage, SyncResultPage, ) -from core.human_input_v2.shared import AccountId, DirectoryScope, IMSyncRunId +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator, IMSyncInitiator +from core.human_input_v2.shared import DirectoryScope, IMSyncRunId from libs.datetime_utils import naive_utc_now from libs.uuid_utils import uuidv7 from models.human_input_v2 import HumanInputIMSyncRun @@ -23,6 +25,7 @@ from repositories.human_input_v2.im_channel_repository import IMChannel from repositories.human_input_v2.im_integration import SQLAlchemyIMControlPlaneRepository from repositories.human_input_v2.sqlalchemy_im_binding_repository import SQLAlchemyIMBindingRepository from repositories.human_input_v2.sqlalchemy_im_identity_repository import SQLAlchemyIMIdentityRepository +from services.account_ports import AccountSnapshotQuery logger = logging.getLogger(__name__) @@ -47,6 +50,18 @@ type IMSyncRunDispatcher = Callable[[IMSyncRunId, DirectoryScope], None] type IMChannelResolver = Callable[[Session, DirectoryScope], IMChannel | None] +@dataclass(frozen=True, slots=True) +class IMSyncRunSummary: + """A persisted run with its currently resolvable Dify Account name. + + The name is empty for EE sys_users and deleted Accounts. CE/SaaS never + resolves EE administrator identities. + """ + + run: IMSyncRun + initiator_display_name: str + + class IMSyncService: """Own Session boundaries around Channel-bound synchronization persistence.""" @@ -56,25 +71,34 @@ class IMSyncService: channel_resolver: IMChannelResolver, dispatcher: IMSyncRunDispatcher, *, + account_query: AccountSnapshotQuery, clock: Callable[[], NaiveDatetime] = naive_utc_now, run_id_factory: Callable[[], IMSyncRunId] = lambda: IMSyncRunId(str(uuidv7())), ) -> None: self._session_factory = session_factory self._channel_resolver = channel_resolver self._dispatcher = dispatcher + self._account_query = account_query self._clock = clock self._run_id_factory = run_id_factory def create_or_get_active_run( self, owner_scope: DirectoryScope, - started_by_account_id: AccountId | None, - ) -> IMSyncRun: + started_by: IMSyncInitiator, + ) -> IMSyncRunSummary: + """Create or reuse a run from one authenticated initiating identity. + + CE/SaaS supplies an Account ID; EE supplies a sys_users.id. Reusing an + active run preserves its original initiator. Commit precedes dispatch; + dispatch failure leaves the queued run available for retry. The returned + display name is resolved from the Account after persistence, never stored. + """ with self._session_factory() as session, session.begin(): channel = self._require_channel(session, owner_scope) decision = SQLAlchemyIMControlPlaneRepository(session, channel).create_or_get_active_run( sync_run_id=self._run_id_factory(), - started_by_account_id=started_by_account_id, + started_by=started_by, now=self._clock(), ) if decision.kind is ActiveRunDecisionKind.STALE_REVISION: @@ -89,9 +113,14 @@ class IMSyncService: raise IMSyncDispatchUnavailableError( "IM synchronization dispatch is temporarily unavailable" ) from error - return decision.run + return self._summarize_run(decision.run) - def get_latest_run(self, owner_scope: DirectoryScope) -> IMSyncRun: + def get_latest_run(self, owner_scope: DirectoryScope) -> IMSyncRunSummary: + """Read the current Channel's latest run and resolve its Account name.""" + return self._summarize_run(self._load_latest_run(owner_scope)) + + def _load_latest_run(self, owner_scope: DirectoryScope) -> IMSyncRun: + """Load the current Channel's latest run or raise the existing not-found error.""" with self._session_factory() as session: channel = self._require_channel(session, owner_scope) run = SQLAlchemyIMControlPlaneRepository(session, channel).load_latest_sync_run() @@ -107,7 +136,7 @@ class IMSyncService: page: int, limit: int, ) -> SyncResultPage: - latest_run = self.get_latest_run(owner_scope) + latest_run = self._load_latest_run(owner_scope) with self._session_factory() as session: channel = self._require_channel(session, owner_scope) return SQLAlchemyIMControlPlaneRepository(session, channel).page_sync_results( @@ -152,6 +181,13 @@ class IMSyncService: raise IMChannelNotConfiguredError("Owner has no IM Channel") return channel + def _summarize_run(self, run: IMSyncRun) -> IMSyncRunSummary: + """Resolve only Account initiators; EE names stay empty.""" + if not isinstance(run.started_by, AccountIMSyncInitiator): + return IMSyncRunSummary(run, "") + account = self._account_query.get(str(run.started_by.account_id)) + return IMSyncRunSummary(run, account.name if account is not None else "") + __all__ = [ "IMChannelNotConfiguredError", diff --git a/api/services/human_input_v2/submission_service.py b/api/services/human_input_v2/submission_service.py index 626065410cf..25304d70bb7 100644 --- a/api/services/human_input_v2/submission_service.py +++ b/api/services/human_input_v2/submission_service.py @@ -11,14 +11,16 @@ from pydantic import JsonValue, TypeAdapter from sqlalchemy.orm import Session, sessionmaker from core.app.file_access import DatabaseFileAccessController +from core.app.file_access.scope import get_current_file_access_scope, grant_upload_file_access from core.human_input_v2.entities import IMProvider from core.human_input_v2.resolved_form import FileInput, FileListInput, MarkdownFragment, ParagraphInput, SelectInput from core.human_input_v2.shared.values import AccountId, TenantId, WorkspaceScope +from core.workflow.file_reference import resolve_file_record_id from core.workflow.nodes.human_input.enums import HumanInputFormStatus from factories.file_factory import build_from_mapping from factories.human_input_v2_form_presentation import refresh_form_file_urls from graphon.enums import WorkflowExecutionStatus -from graphon.file import FileUploadConfig +from graphon.file import FileTransferMethod, FileUploadConfig from graphon.variables.factory import build_segment from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment, StringSegment from libs.datetime_utils import naive_utc_now @@ -144,6 +146,41 @@ class HumanInputSubmissionService: def __init__(self, *, session_factory: sessionmaker[Session]) -> None: self._session_factory = session_factory + def grant_submitted_file_access(self, *, app_id: str, workflow_run_id: str) -> None: + """Restore accepted HITL file grants inside the initiating EndUser's scope. + + The caller supplies its authorized app and run before worker contexts + are captured. Grant only persisted submissions in the current tenant, + including earlier HITL steps across repeated pauses. Keep persistent + file ownership and all unrelated grants unchanged. Account and absent + scopes require no grants and perform no reads; read failures propagate + without changing the scope. + """ + scope = get_current_file_access_scope() + if scope is None or not scope.requires_user_ownership: + return + + with self._session_factory() as session: + forms = SQLAlchemyFormRepository(session, TenantId(scope.tenant_id), app_id).get_submitted_forms( + workflow_run_id=workflow_run_id + ) + file_ids: set[str] = set() + for form in forms: + assert form.submission is not None, "Submitted form must contain its accepted submission" + for segment in form.submission.normalized_submission_data.values(): + if isinstance(segment, FileSegment): + files = [segment.value] + elif isinstance(segment, ArrayFileSegment): + files = segment.value + else: + files = [] + for file in files: + if file.transfer_method in {FileTransferMethod.LOCAL_FILE, FileTransferMethod.REMOTE_URL}: + file_id = resolve_file_record_id(file.reference) + if file_id is not None: + file_ids.add(file_id) + grant_upload_file_access(file_ids) + def get_service_api_form( self, *, actor: ServiceAPIFormActor, form_token: str, ip_address: str | None = None ) -> Form: diff --git a/api/tasks/app_generate/workflow_execute_task.py b/api/tasks/app_generate/workflow_execute_task.py index 6b47a01beab..963e0e996c2 100644 --- a/api/tasks/app_generate/workflow_execute_task.py +++ b/api/tasks/app_generate/workflow_execute_task.py @@ -40,6 +40,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.human_input_v2.submission_service import HumanInputSubmissionService from services.workflow_execution_identity import build_workflow_execution_identity from services.workflow_run_agg import WorkflowRunAgg @@ -715,6 +716,11 @@ def _resume_advanced_chat( graph_runtime_state=graph_runtime_state, pause_state_config=pause_state_config, response_stream_filter=response_stream_filter, + restore_file_access=partial( + HumanInputSubmissionService(session_factory=session_factory).grant_submitted_file_access, + app_id=app_model.id, + workflow_run_id=workflow_run_id, + ), session=session, ) except Exception: @@ -782,6 +788,11 @@ def _resume_workflow( workflow_node_execution_repository=workflow_node_execution_repository, pause_state_config=pause_state_config, response_stream_filter=response_stream_filter, + restore_file_access=partial( + HumanInputSubmissionService(session_factory=session_factory).grant_submitted_file_access, + app_id=app_model.id, + workflow_run_id=workflow_run_id, + ), ) except Exception: logger.exception("Failed to resume workflow execution for workflow run %s", workflow_run_id) diff --git a/api/tasks/async_workflow_tasks.py b/api/tasks/async_workflow_tasks.py index a9e3890e915..63616064c99 100644 --- a/api/tasks/async_workflow_tasks.py +++ b/api/tasks/async_workflow_tasks.py @@ -7,6 +7,7 @@ with appropriate retry policies and error handling. import logging from datetime import UTC, datetime +from functools import partial from typing import Any from uuid import UUID @@ -30,6 +31,7 @@ from models.workflow import Workflow, WorkflowNodeExecutionTriggeredFrom, Workfl from repositories.factory import DifyAPIRepositoryFactory from services.app_task_service import AppTaskService from services.errors.app import WorkflowNotFoundError +from services.human_input_v2.submission_service import HumanInputSubmissionService from services.trigger.execution_layer import TriggerPostLayer from services.workflow.entities import ( WorkflowResumeTaskData, @@ -184,6 +186,11 @@ def resume_workflow_execution(task_data_dict: dict[str, Any]) -> None: graph_engine_layers=graph_engine_layers, pause_state_config=pause_config, response_stream_filter=response_stream_filter, + restore_file_access=partial( + HumanInputSubmissionService(session_factory=session_factory).grant_submitted_file_access, + app_id=app_model.id, + workflow_run_id=workflow_run.id, + ), ) workflow_run_repo.delete_workflow_pause(pause_entity) diff --git a/api/tests/e2e_tests/containers/controllers/console/workspace/test_im_contact_sync_api.py b/api/tests/e2e_tests/containers/controllers/console/workspace/test_im_contact_sync_api.py index 5fde53b991a..6d61c26d585 100644 --- a/api/tests/e2e_tests/containers/controllers/console/workspace/test_im_contact_sync_api.py +++ b/api/tests/e2e_tests/containers/controllers/console/workspace/test_im_contact_sync_api.py @@ -23,6 +23,7 @@ from core.human_input_v2.im_integration import ( SyncIdentitySnapshot, SyncResultFact, ) +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator from core.human_input_v2.shared import ( AccountId, ContactId, @@ -275,7 +276,7 @@ def test_latest_result_http_projection_reads_every_persisted_bucket( sync_run_id=IMSyncRunId("00000000-0000-0000-0000-000000000501"), channel_revision=IMChannelRevision(str(channel.id), channel.config_version), provider=channel.provider, - started_by_account_id=AccountId(account.id), + started_by=AccountIMSyncInitiator(AccountId(account.id)), now=_NOW, ), status=IMSyncRunStatus.SUCCEEDED, diff --git a/api/tests/integration_tests/repositories/human_input_v2/test_im_control_plane_concurrency.py b/api/tests/integration_tests/repositories/human_input_v2/test_im_control_plane_concurrency.py index a4a1ad7f5ba..47b3ede3a3a 100644 --- a/api/tests/integration_tests/repositories/human_input_v2/test_im_control_plane_concurrency.py +++ b/api/tests/integration_tests/repositories/human_input_v2/test_im_control_plane_concurrency.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session, sessionmaker from core.human_input_v2.entities import IMProvider from core.human_input_v2.im_integration import ActiveRunDecisionKind +from core.human_input_v2.im_integration.sync_records import SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import ContactId, IMIdentityId, IMSyncRunId from extensions.ext_database import db from libs.datetime_utils import naive_utc_now @@ -99,7 +100,7 @@ def test_concurrent_sync_triggers_create_at_most_one_active_run(flask_req_ctx: o with session_maker.begin() as session: decision = SQLAlchemyIMControlPlaneRepository(session, channel).create_or_get_active_run( sync_run_id=IMSyncRunId(str(uuidv7())), - started_by_account_id=None, + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), now=naive_utc_now(), ) return decision.kind diff --git a/api/tests/integration_tests/repositories/human_input_v2/test_stale_im_sync_runs.py b/api/tests/integration_tests/repositories/human_input_v2/test_stale_im_sync_runs.py index 7e5cc2cfd06..c63b38bac80 100644 --- a/api/tests/integration_tests/repositories/human_input_v2/test_stale_im_sync_runs.py +++ b/api/tests/integration_tests/repositories/human_input_v2/test_stale_im_sync_runs.py @@ -30,6 +30,7 @@ from core.human_input_v2.im_integration.adapters import ( IMWebhookHandler, ProviderUserId, ) +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator, SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import AccountId, ContactId, DirectoryScope, IMSyncRunId, TenantId, WorkspaceScope from libs.datetime_utils import naive_utc_now from models.human_input_v2 import ( @@ -41,6 +42,7 @@ from models.human_input_v2 import ( HumanInputIMSyncResult, HumanInputIMSyncRun, ) +from repositories.account.repository import SQLAlchemyAccountRepository from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials from repositories.human_input_v2.contact import Contact from repositories.human_input_v2.im_channel_repository import IMChannel, IMChannelId, IMChannelStatus, WebhookId @@ -151,7 +153,12 @@ class _SyncContext: return WorkspaceIMChannelReader(session, self.scope.id).get() def service(self) -> IMSyncService: - return IMSyncService(self.sessions, self.resolve_channel, lambda _run, _scope: None) + return IMSyncService( + self.sessions, + self.resolve_channel, + lambda _run, _scope: None, + account_query=SQLAlchemyAccountRepository(self.sessions), + ) def worker(self, directory: _Directory | None = None) -> IMContactSyncWorker: def reconciliation(scope: DirectoryScope) -> IMChannelReconciliationService: @@ -201,7 +208,7 @@ def sync_context(sync_engine: Engine) -> _SyncContext: def test_revision_rotation_retires_old_run_and_new_worker_succeeds( sync_context: _SyncContext, status: IMSyncRunStatus, recovery: str ) -> None: - old = sync_context.service().create_or_get_active_run(sync_context.scope, None) + old = sync_context.service().create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)).run with sync_context.sessions.begin() as session: session.get_one(HumanInputIMSyncRun, str(old.id)).status = status sync_context.rotate() @@ -209,7 +216,9 @@ def test_revision_rotation_retires_old_run_and_new_worker_succeeds( terminal = sync_context.worker().execute(old.id, sync_context.scope) assert terminal.status is IMSyncRunStatus.FAILED assert terminal.error_code == "stale_revision" - current = sync_context.service().create_or_get_active_run(sync_context.scope, None) + current = ( + sync_context.service().create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)).run + ) assert current.id != old.id assert current.channel_revision.config_version == 2 assert current.status is IMSyncRunStatus.QUEUED @@ -233,14 +242,18 @@ def test_revision_rotation_retires_old_run_and_new_worker_succeeds( def test_concurrent_old_workers_and_triggers_leave_one_new_active_run(sync_context: _SyncContext) -> None: - old = sync_context.service().create_or_get_active_run(sync_context.scope, None) + old = sync_context.service().create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)).run sync_context.rotate() barrier = Barrier(4) def execute(trigger: bool) -> IMSyncRun: barrier.wait(timeout=10) if trigger: - return sync_context.service().create_or_get_active_run(sync_context.scope, None) + return ( + sync_context.service() + .create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)) + .run + ) return sync_context.worker().execute(old.id, sync_context.scope) with ThreadPoolExecutor(max_workers=4) as executor: @@ -267,14 +280,18 @@ def test_concurrent_old_workers_and_triggers_leave_one_new_active_run(sync_conte def test_late_old_apply_cannot_overwrite_new_revision_state(sync_context: _SyncContext) -> None: - old = sync_context.service().create_or_get_active_run(sync_context.scope, None) + old = sync_context.service().create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)).run entered, resume = Event(), Event() with ThreadPoolExecutor(max_workers=1) as executor: pending = executor.submit(sync_context.worker(_Directory(entered, resume)).execute, old.id, sync_context.scope) try: assert entered.wait(timeout=10) sync_context.rotate() - current = sync_context.service().create_or_get_active_run(sync_context.scope, None) + current = ( + sync_context.service() + .create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)) + .run + ) assert sync_context.worker().execute(current.id, sync_context.scope).status is IMSyncRunStatus.SUCCEEDED finally: resume.set() @@ -296,7 +313,7 @@ def test_late_old_apply_cannot_overwrite_new_revision_state(sync_context: _SyncC def test_stale_completion_is_scoped_current_and_caller_rollback_owned(sync_context: _SyncContext) -> None: - old = sync_context.service().create_or_get_active_run(sync_context.scope, None) + old = sync_context.service().create_or_get_active_run(sync_context.scope, AccountIMSyncInitiator(_ACCOUNT_ID)).run current = sync_context.rotate() with sync_context.sessions() as session: assert ( @@ -307,7 +324,9 @@ def test_stale_completion_is_scoped_current_and_caller_rollback_owned(sync_conte ) with sync_context.sessions() as session: replacement = SQLAlchemyIMControlPlaneRepository(session, current).create_or_get_active_run( - sync_run_id=IMSyncRunId(str(uuid4())), started_by_account_id=None, now=naive_utc_now() + sync_run_id=IMSyncRunId(str(uuid4())), + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), + now=naive_utc_now(), ) assert replacement.run is not None assert replacement.run.id != old.id diff --git a/api/tests/unit_tests/controllers/common/test_human_input_v2_contracts.py b/api/tests/unit_tests/controllers/common/test_human_input_v2_contracts.py index 6168adf7d5e..fc5c67b4294 100644 --- a/api/tests/unit_tests/controllers/common/test_human_input_v2_contracts.py +++ b/api/tests/unit_tests/controllers/common/test_human_input_v2_contracts.py @@ -39,6 +39,7 @@ def test_sync_run_exposes_captured_revision() -> None: "provider": "feishu", "channel_id": "channel-id", "channel_config_version": 4, + "initiator_display_name": "Sync Admin", } ) diff --git a/api/tests/unit_tests/controllers/console/workspace/test_im_contact_sync_api.py b/api/tests/unit_tests/controllers/console/workspace/test_im_contact_sync_api.py index 3d60c4c6c79..2c51914054c 100644 --- a/api/tests/unit_tests/controllers/console/workspace/test_im_contact_sync_api.py +++ b/api/tests/unit_tests/controllers/console/workspace/test_im_contact_sync_api.py @@ -7,7 +7,8 @@ from dataclasses import replace from datetime import datetime from importlib import import_module from inspect import unwrap -from typing import Never, override +from typing import override +from unittest.mock import create_autospec import pytest from flask import Flask @@ -50,10 +51,12 @@ from core.human_input_v2.im_integration import ( SyncResultFact, SyncResultPage, ) +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator, IMSyncInitiator from core.human_input_v2.shared import ( AccountId, ContactId, DeploymentScope, + DirectoryScope, IMBindingId, IMIdentityId, IMSyncResultId, @@ -77,6 +80,7 @@ from services.human_input_v2.im_contact_sync.service import ( IMSyncDispatchUnavailableError, IMSyncRevisionChangedError, IMSyncRunNotFoundError, + IMSyncRunSummary, ) from tests.helpers.unit.model_factories import make_account @@ -87,7 +91,7 @@ _RUN = IMSyncRun.create( sync_run_id=IMSyncRunId("run-1"), channel_revision=IMChannelRevision("channel-1", 3), provider=IMProvider.FEISHU, - started_by_account_id=AccountId("account-1"), + started_by=AccountIMSyncInitiator(AccountId("account-1")), now=_NOW, ) _RESULT = SyncResultFact( @@ -117,39 +121,39 @@ _RESULT = SyncResultFact( class _SyncService: def create_or_get_active_run( self, - organization_scope: WorkspaceScope, - started_by_account_id: AccountId, - ) -> IMSyncRun: - assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) - assert started_by_account_id == AccountId("account-1") - return _RUN + owner_scope: DirectoryScope, + started_by: IMSyncInitiator, + ) -> IMSyncRunSummary: + assert owner_scope == WorkspaceScope(id=TenantId("workspace-1")) + assert started_by == AccountIMSyncInitiator(AccountId("account-1")) + return IMSyncRunSummary(_RUN, "Editor") - def get_latest_run(self, organization_scope: WorkspaceScope) -> IMSyncRun: - assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) - return _RUN + def get_latest_run(self, owner_scope: DirectoryScope) -> IMSyncRunSummary: + assert owner_scope == WorkspaceScope(id=TenantId("workspace-1")) + return IMSyncRunSummary(_RUN, "Editor") def list_latest_results( self, - organization_scope: WorkspaceScope, + owner_scope: DirectoryScope, result_type: IMSyncResultType, *, page: int, limit: int, ) -> SyncResultPage: - assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) + assert owner_scope == WorkspaceScope(id=TenantId("workspace-1")) assert result_type is IMSyncResultType.NOT_MATCHED assert (page, limit) == (1, 20) return SyncResultPage((_RESULT,), page=page, limit=limit, total=1) def search_identities( self, - organization_scope: WorkspaceScope, + owner_scope: DirectoryScope, *, keyword: str | None, page: int, limit: int, ) -> SynchronizedIMIdentityPage: - assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) + assert owner_scope == WorkspaceScope(id=TenantId("workspace-1")) assert (keyword, page, limit) == ("provider-user", 1, 20) return SynchronizedIMIdentityPage( ( @@ -188,26 +192,42 @@ def _contact_view(scope: IMBindingScope) -> ContactIMBindingView: class _BindingService: - def set_organization_binding(self, **kwargs: object) -> ContactIMBindingView: - assert kwargs == { - "organization_scope": WorkspaceScope(id=TenantId("workspace-1")), - "tenant_id": TenantId("workspace-1"), - "contact_id": ContactId("00000000-0000-0000-0000-000000000001"), - "identity_id": IMIdentityId("identity-1"), - "bound_by_account_id": AccountId("account-1"), - } + def set_organization_binding( + self, + *, + organization_scope: DirectoryScope, + tenant_id: TenantId, + contact_id: ContactId, + identity_id: IMIdentityId, + bound_by_account_id: AccountId | None, + ) -> ContactIMBindingView: + assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) + assert tenant_id == TenantId("workspace-1") + assert contact_id == ContactId("00000000-0000-0000-0000-000000000001") + assert identity_id == IMIdentityId("identity-1") + assert bound_by_account_id == AccountId("account-1") return _contact_view(IMBindingScope.ORGANIZATION) - def delete_organization_binding(self, **kwargs: object) -> None: - assert kwargs == { - "organization_scope": WorkspaceScope(id=TenantId("workspace-1")), - "contact_id": ContactId("00000000-0000-0000-0000-000000000001"), - } + def delete_organization_binding(self, *, organization_scope: DirectoryScope, contact_id: ContactId) -> None: + assert organization_scope == WorkspaceScope(id=TenantId("workspace-1")) + assert contact_id == ContactId("00000000-0000-0000-0000-000000000001") - def set_workspace_override(self, **_kwargs: object) -> ContactIMBindingView: + def set_workspace_override( + self, + *, + organization_scope: DirectoryScope, + tenant_id: TenantId, + contact_id: ContactId, + identity_id: IMIdentityId, + bound_by_account_id: AccountId | None, + ) -> ContactIMBindingView: + del organization_scope, tenant_id, contact_id, identity_id, bound_by_account_id return _contact_view(IMBindingScope.WORKSPACE) - def reset_workspace_override(self, **_kwargs: object) -> ContactIMBindingView: + def reset_workspace_override( + self, *, organization_scope: DirectoryScope, tenant_id: TenantId, contact_id: ContactId + ) -> ContactIMBindingView: + del organization_scope, tenant_id, contact_id return _contact_view(IMBindingScope.ORGANIZATION) @@ -265,6 +285,7 @@ def test_sync_command_and_queries_map_transport_neutral_results( "provider": "feishu", "channel_id": "channel-1", "channel_config_version": 3, + "initiator_display_name": "Editor", } } ) @@ -310,11 +331,11 @@ def test_sync_dispatch_unavailable_has_stable_retryable_http_mapping( ) -> None: error = IMSyncDispatchUnavailableError("IM synchronization dispatch is temporarily unavailable") - class SyncService: - def create_or_get_active_run(self, _scope: object, _account_id: object) -> Never: - raise error - - monkeypatch.setattr(application.sync_service, "create_or_get_active_run", SyncService().create_or_get_active_run) + monkeypatch.setattr( + application.sync_service, + "create_or_get_active_run", + create_autospec(application.sync_service.create_or_get_active_run, side_effect=error), + ) with app.test_request_context(method="POST"): response, status = unwrap(WorkspaceIMSyncRunsApi.post)( @@ -346,22 +367,21 @@ def test_write_unavailable_has_stable_retryable_http_mapping( ) error = IMWriteUnavailableError("IM write is temporarily unavailable") - class SyncService: - def create_or_get_active_run(self, _scope: object, _account_id: object) -> Never: - raise error - - class BindingService: - def set_organization_binding(self, **_kwargs: object) -> Never: - raise error - - def set_workspace_override(self, **_kwargs: object) -> Never: - raise error - - monkeypatch.setattr(application.sync_service, "create_or_get_active_run", SyncService().create_or_get_active_run) monkeypatch.setattr( - application.binding_service, "set_organization_binding", BindingService().set_organization_binding + application.sync_service, + "create_or_get_active_run", + create_autospec(application.sync_service.create_or_get_active_run, side_effect=error), + ) + monkeypatch.setattr( + application.binding_service, + "set_organization_binding", + create_autospec(application.binding_service.set_organization_binding, side_effect=error), + ) + monkeypatch.setattr( + application.binding_service, + "set_workspace_override", + create_autospec(application.binding_service.set_workspace_override, side_effect=error), ) - monkeypatch.setattr(application.binding_service, "set_workspace_override", BindingService().set_workspace_override) account = make_account(account_id="account-1", name="Editor", email="editor@example.com") contact_id = "00000000-0000-0000-0000-000000000001" if command == "sync_create": @@ -399,7 +419,7 @@ def test_write_unavailable_has_stable_retryable_http_mapping( contact_snapshot=SyncContactSnapshot( ContactId("contact-1"), "Reviewer", - None, + "snapshot@example.com", None, _CONTACT_CREATED_AT, ), @@ -410,7 +430,7 @@ def test_write_unavailable_has_stable_retryable_http_mapping( contact_id=ContactId("contact-1"), binding_id=IMBindingId("binding-1"), removal_reason=IMSyncRemovalReason.NOT_PRESENT_IN_DIRECTORY, - contact_snapshot=SyncContactSnapshot(ContactId("contact-1"), "Reviewer", None, None), + contact_snapshot=SyncContactSnapshot(ContactId("contact-1"), "Reviewer", "snapshot@example.com", None), identity_snapshot=SyncIdentitySnapshot( IMIdentityId("identity-1"), IMProvider.FEISHU, @@ -435,10 +455,11 @@ def test_latest_results_map_every_discriminated_bucket( monkeypatch: pytest.MonkeyPatch, sync_result: SyncResultFact, ) -> None: - class SyncService: + class SyncService(_SyncService): + @override def list_latest_results( self, - _scope: object, + owner_scope: DirectoryScope, result_type: IMSyncResultType, *, page: int, @@ -461,6 +482,8 @@ def test_latest_results_map_every_discriminated_bucket( assert result_payload["type"] == sync_result.result_type.value if sync_result.result_type in (IMSyncResultType.ADDED, IMSyncResultType.SKIPPED, IMSyncResultType.REMOVED): assert "type" not in result_payload["contact"] + assert sync_result.contact_snapshot is not None + assert result_payload["contact"]["email"] == sync_result.contact_snapshot.email if sync_result.result_type is IMSyncResultType.ADDED: assert result_payload["contact"]["created_at"] == 1752127200 if sync_result.result_type is IMSyncResultType.REMOVED: @@ -569,25 +592,50 @@ def test_enterprise_override_and_identity_search_use_the_deployment_directory( config_overrides: Callable[..., None], ) -> None: config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.ENTERPRISE) - scopes = [] + scopes: list[DirectoryScope] = [] class BindingService(_BindingService): @override - def set_workspace_override(self, **kwargs: object) -> ContactIMBindingView: - scopes.append(kwargs["organization_scope"]) - return super().set_workspace_override(**kwargs) + def set_workspace_override( + self, + *, + organization_scope: DirectoryScope, + tenant_id: TenantId, + contact_id: ContactId, + identity_id: IMIdentityId, + bound_by_account_id: AccountId | None, + ) -> ContactIMBindingView: + scopes.append(organization_scope) + return super().set_workspace_override( + organization_scope=organization_scope, + tenant_id=tenant_id, + contact_id=contact_id, + identity_id=identity_id, + bound_by_account_id=bound_by_account_id, + ) @override - def reset_workspace_override(self, **kwargs: object) -> ContactIMBindingView: - scopes.append(kwargs["organization_scope"]) - return super().reset_workspace_override(**kwargs) + def reset_workspace_override( + self, *, organization_scope: DirectoryScope, tenant_id: TenantId, contact_id: ContactId + ) -> ContactIMBindingView: + scopes.append(organization_scope) + return super().reset_workspace_override( + organization_scope=organization_scope, + tenant_id=tenant_id, + contact_id=contact_id, + ) class SyncService(_SyncService): @override def search_identities( - self, organization_scope: WorkspaceScope, **_kwargs: object + self, + owner_scope: DirectoryScope, + *, + keyword: str | None, + page: int, + limit: int, ) -> SynchronizedIMIdentityPage: - scopes.append(organization_scope) + scopes.append(owner_scope) return SynchronizedIMIdentityPage((), page=1, limit=20, total=0) monkeypatch.setattr(application.binding_service, "set_workspace_override", BindingService().set_workspace_override) @@ -638,13 +686,10 @@ def test_expected_application_errors_have_stable_http_mapping( expected_code: str, ) -> None: if isinstance(error, IMBindingCommandError): - - class BindingService: - def set_organization_binding(self, **_kwargs: object) -> Never: - raise error - monkeypatch.setattr( - application.binding_service, "set_organization_binding", BindingService().set_organization_binding + application.binding_service, + "set_organization_binding", + create_autospec(application.binding_service.set_organization_binding, side_effect=error), ) handler = unwrap(WorkspaceContactIMBindingsApi.put) request_context = app.test_request_context(method="PUT", json={"identity_id": "identity-1"}) @@ -655,13 +700,10 @@ def test_expected_application_errors_have_stable_http_mapping( "00000000-0000-0000-0000-000000000001", ) elif isinstance(error, IMSyncRevisionChangedError): - - class SyncService: - def create_or_get_active_run(self, _scope: object, _account_id: object) -> Never: - raise error - monkeypatch.setattr( - application.sync_service, "create_or_get_active_run", SyncService().create_or_get_active_run + application.sync_service, + "create_or_get_active_run", + create_autospec(application.sync_service.create_or_get_active_run, side_effect=error), ) handler = unwrap(WorkspaceIMSyncRunsApi.post) request_context = app.test_request_context(method="POST") @@ -671,12 +713,11 @@ def test_expected_application_errors_have_stable_http_mapping( make_account(account_id="account-1", name="Editor", email="editor@example.com"), ) else: - - class SyncService: - def get_latest_run(self, _scope: object) -> Never: - raise error - - monkeypatch.setattr(application.sync_service, "get_latest_run", SyncService().get_latest_run) + monkeypatch.setattr( + application.sync_service, + "get_latest_run", + create_autospec(application.sync_service.get_latest_run, side_effect=error), + ) handler = unwrap(WorkspaceLatestIMSyncRunApi.get) request_context = app.test_request_context(method="GET") handler_args = (WorkspaceLatestIMSyncRunApi(), "workspace-1") @@ -823,7 +864,7 @@ def test_contact_read_handlers_preserve_current_and_editor_safe_projections( assert options["data"][0]["im_providers"] == ["feishu"] assert batch_options["data"] == options["data"] assert "im_bindings" not in options["data"][0] - assert "email" not in batch["data"][0] + assert batch["data"][0]["email"] == "reviewer@example.com" @pytest.mark.parametrize("provider", list(IMProvider)) diff --git a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py b/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py index ed8a48ee0c4..bfc0db4e76e 100644 --- a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py +++ b/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py @@ -5,7 +5,9 @@ import logging import time from collections.abc import Mapping from contextlib import contextmanager +from datetime import timedelta from decimal import Decimal +from functools import partial from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock @@ -18,7 +20,7 @@ from pytest_mock import MockerFixture from sqlalchemy import Engine as _app_generator_Engine from sqlalchemy import event as sqlalchemy_event from sqlalchemy.engine import Engine as _advanced_chat_app_generator_Engine -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker import core.app.apps.advanced_chat.app_generator as app_generator_module import core.workflow.nodes.human_input.entities # noqa: F401 @@ -31,6 +33,7 @@ from core.app.apps.advanced_chat.generate_task_pipeline import ( MessageSnapshot, WorkflowSnapshot, ) +from core.app.apps.base_app_generator import noop_restore_file_access from core.app.apps.exc import GenerateTaskStoppedError from core.app.apps.workflow import app_generator as wf_app_gen_module from core.app.entities.app_invoke_entities import ( @@ -43,7 +46,10 @@ from core.app.file_access import DatabaseFileAccessController from core.app.task_pipeline import message_cycle_manager from core.app.task_pipeline.message_cycle_manager import MessageCycleManager from core.app.workflow.file_runtime import DifyWorkflowFileRuntime +from core.human_input_v2.resolved_form import ResolvedForm +from core.human_input_v2.shared.values import RecipientId, TenantId from core.workflow import node_factory as node_factory_module +from core.workflow.file_reference import build_file_reference from core.workflow.node_factory import DifyNodeFactory from core.workflow.system_variables import build_system_variables from graphon.engine import Engine as _pause_resume_Engine @@ -59,6 +65,7 @@ from graphon.entities import WorkflowStartReason from graphon.entities.base_node_data import BaseNodeData, RetryConfig from graphon.entities.pause_reason import SchedulingPause from graphon.enums import BuiltinNodeTypes, NodeType, WorkflowNodeExecutionStatus +from graphon.file import File, FileTransferMethod, FileType from graphon.graph import Graph from graphon.node_events import NodeRunResult, PauseRequestedEvent from graphon.nodes.base.entities import OutputVariableEntity @@ -67,6 +74,7 @@ from graphon.nodes.end.entities import EndNodeData from graphon.nodes.start.entities import StartNodeData from graphon.runtime import RuntimeState, VariablePool from graphon.variables.input_entities import VariableEntity +from graphon.variables.segments import FileSegment from libs.datetime_utils import naive_utc_now from models import Account as _advanced_chat_app_generator_Account from models import Workflow as _advanced_chat_app_generator_Workflow @@ -77,7 +85,10 @@ from models.model import App, AppMode, Conversation, EndUser, Message from models.workflow import Workflow as _app_generator_Workflow from models.workflow import Workflow as _pause_resume_Workflow from models.workflow import WorkflowType +from repositories.human_input_v2.form_repository import Form, FormCreateParams +from repositories.human_input_v2.sqlalchemy_form_repository import SQLAlchemyFormRepository from services.errors.conversation import ConversationNotExistsError +from services.human_input_v2.submission_service import HumanInputSubmissionService from services.workflow_run_agg import WorkflowRunAgg from tests.helpers.unit.config_override import apply_config_overrides from tests.helpers.unit.model_factories import make_upload_file @@ -671,9 +682,15 @@ class TestAdvancedChatAppGeneratorInternals: assert isinstance(captured["conversation"], ConversationSnapshot) assert isinstance(captured["message"], MessageSnapshot) + @pytest.mark.parametrize("resuming", [False, True], ids=["file-default", "submitted-hitl-file"]) def test_generate_internal_flow_with_existing_records_skips_init( - self, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session, sqlite_engine: _app_generator_Engine - ): + self, + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + sqlite_engine: _app_generator_Engine, + resuming: bool, + sqlite_session_factory: sessionmaker[Session], + ) -> None: generator = AdvancedChatAppGenerator(execution_driver=WorkflowRunAgg.run) generator._dialogue_count = 0 app_config = self._build_app_config() @@ -688,6 +705,40 @@ class TestAdvancedChatAppGeneratorInternals: } ) ] + if resuming: + app_config.variables = [] + now = naive_utc_now() + repository = SQLAlchemyFormRepository(sqlite_session, TenantId(app_config.tenant_id), app_config.app_id) + form = repository.create_form( + FormCreateParams( + workflow_run_id="run-id", + node_execution_id="approval-execution", + resolved_form=ResolvedForm(title="Approval", parts=(), actions=(), legacy_form_content=""), + expiration_time=now + timedelta(hours=1), + global_timeout_deadline=now + timedelta(days=1), + ) + ) + accepted = repository.submit_form( + form.id, + submitted_by_recipient_id=RecipientId("recipient"), + submitted_by_delivery_id="delivery", + selected_action_id="approve", + raw_submission_inputs={}, + normalized_submission_data={ + "file": FileSegment( + value=File( + file_type=FileType.DOCUMENT, + transfer_method=FileTransferMethod.LOCAL_FILE, + reference=build_file_reference(record_id=default_file.id), + filename=default_file.name, + extension=".txt", + mime_type=default_file.mime_type, + size=default_file.size, + ) + ) + }, + ) + assert isinstance(accepted, Form) file_runtime = DifyWorkflowFileRuntime(file_access_controller=DatabaseFileAccessController()) monkeypatch.setattr(file_runtime, "_secret_key", lambda: b"test-key") @@ -780,6 +831,16 @@ class TestAdvancedChatAppGeneratorInternals: conversation=conversation, message=message, stream=False, + graph_runtime_state=_build_runtime_state("run-id") if resuming else None, + restore_file_access=( + partial( + HumanInputSubmissionService(session_factory=sqlite_session_factory).grant_submitted_file_access, + app_id=app_config.app_id, + workflow_run_id="run-id", + ) + if resuming + else noop_restore_file_access + ), ) assert response == {"raw": True} 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 1b8896b29d2..51cc7dc942e 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 @@ -3,6 +3,8 @@ from __future__ import annotations import contextlib import contextvars import json +from datetime import timedelta +from functools import partial from types import SimpleNamespace from unittest.mock import MagicMock, Mock @@ -31,22 +33,31 @@ from core.app.layers.pause_state_persist_layer import ( _WorkflowGenerateEntityWrapper, ) from core.app.workflow.file_runtime import DifyWorkflowFileRuntime +from core.human_input_v2.resolved_form import ResolvedForm +from core.human_input_v2.shared.values import RecipientId, TenantId 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.engine_events import GraphRunSucceededEvent from graphon.enums import WorkflowExecutionStatus +from graphon.file import File, FileTransferMethod, FileType from graphon.runtime import RuntimeState, VariablePool from graphon.variable_loader import DUMMY_VARIABLE_LOADER from graphon.variables.input_entities import VariableEntity +from graphon.variables.segments import ArrayFileSegment, FileSegment +from libs.datetime_utils import naive_utc_now 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 repositories.human_input_v2.form_repository import Form, FormCreateParams +from repositories.human_input_v2.sqlalchemy_form_repository import SQLAlchemyFormRepository from services.file_request_service import FileRequestService +from services.human_input_v2.submission_service import HumanInputSubmissionService from services.workflow_run_agg import WorkflowRunAgg from tests.helpers.unit.model_factories import make_account, make_upload_file, make_workflow @@ -195,6 +206,140 @@ class TestWorkflowAppGeneratorCases: def disable_enterprise_telemetry(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("core.telemetry.gateway.is_enterprise_telemetry_enabled", lambda: False) + @staticmethod + @pytest.mark.parametrize("file_list", [False, True], ids=["single-file", "file-list"]) + @pytest.mark.parametrize("invoke_from", [InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API]) + def test_resumed_end_user_can_output_accepted_hitl_files( + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + sqlite_session_factory: sessionmaker[Session], + file_list: bool, + invoke_from: InvokeFrom, + ) -> None: + app, workflow, end_user = _persist_generator_rows(sqlite_session) + uploaded = make_upload_file(tenant_id=app.tenant_id, created_by="approver") + unrelated = make_upload_file(tenant_id=app.tenant_id, created_by="approver") + workflow.graph = json.dumps( + { + "nodes": [ + {"id": "start", "data": {"type": "start", "title": "Start", "variables": []}}, + { + "id": "end", + "data": { + "type": "end", + "title": "End", + "outputs": [{"variable": "result", "value_selector": ["approval", "file"]}], + }, + }, + ], + "edges": [{"source": "start", "target": "end"}], + } + ) + sqlite_session.add_all( + [ + uploaded, + unrelated, + WorkflowRun( + id="run", + tenant_id=app.tenant_id, + app_id=app.id, + workflow_id=workflow.id, + type=workflow.type, + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + version=workflow.version, + graph=workflow.graph, + inputs="{}", + status=WorkflowExecutionStatus.PAUSED, + created_by_role=CreatorUserRole.END_USER, + created_by=end_user.id, + ), + ] + ) + sqlite_session.commit() + file = File( + file_type=FileType.DOCUMENT, + transfer_method=FileTransferMethod.LOCAL_FILE, + reference=build_file_reference(record_id=uploaded.id), + filename=uploaded.name, + extension=".txt", + mime_type=uploaded.mime_type, + size=uploaded.size, + ) + submitted = ArrayFileSegment(value=[file]) if file_list else FileSegment(value=file) + now = naive_utc_now() + with sqlite_session_factory.begin() as session: + repository = SQLAlchemyFormRepository(session, TenantId(app.tenant_id), app.id) + form = repository.create_form( + FormCreateParams( + workflow_run_id="run", + node_execution_id="approval-execution", + resolved_form=ResolvedForm(title="Approval", parts=(), actions=(), legacy_form_content=""), + expiration_time=now + timedelta(hours=1), + global_timeout_deadline=now + timedelta(days=1), + ) + ) + accepted = repository.submit_form( + form.id, + submitted_by_recipient_id=RecipientId("recipient"), + submitted_by_delivery_id="delivery", + selected_action_id="approve", + raw_submission_inputs={}, + normalized_submission_data={"file": submitted}, + ) + assert isinstance(accepted, Form) + state = RuntimeState(workflow_id=workflow.id, variable_pool=VariablePool(), start_at=0) + state.variable_pool.add(["approval", "file"], submitted) + state.variable_pool.add(["sys", "workflow_run_id"], "run") + state.variable_pool.add(["sys", "app_id"], app.id) + file_runtime = DifyWorkflowFileRuntime(file_access_controller=DatabaseFileAccessController()) + monkeypatch.setattr(file_runtime, "_secret_key", lambda: b"test-key") + completed: list[GraphRunSucceededEvent] = [] + + def execute( + runner: WorkflowBasedAppRunner, + _pause_config: PauseStateLayerConfig | None, + _executions: WorkflowExecutionRepository, + _nodes: WorkflowNodeExecutionRepository, + ) -> None: + prepared = runner.prepare() + assert prepared is not None + events = list(prepared.entry.run()) + completed.extend(event for event in events if isinstance(event, GraphRunSucceededEvent)) + + def respond(**_kwargs: object) -> dict[str, object]: + assert file_runtime.resolve_upload_file_uri(upload_file_id=uploaded.id).startswith( + f"/files/{uploaded.id}/file-preview?" + ) + with pytest.raises(ValueError, match="not found"): + file_runtime.resolve_upload_file_uri(upload_file_id=unrelated.id) + return {} + + generator = WorkflowAppGenerator(execution_driver=execute) + monkeypatch.setattr(generator, "_handle_response", respond) + monkeypatch.setattr( + _workflow_app_generator.WorkflowAppGenerateResponseConverter, + "convert", + lambda response, invoke_from: response, + ) + execution_repository, node_repository = _repositories(sqlite_session_factory, app, end_user) + generator.resume( + app_model=app, + workflow=workflow, + user=end_user, + application_generate_entity=_generate_entity(app, workflow, end_user, invoke_from=invoke_from), + graph_runtime_state=state, + workflow_execution_repository=execution_repository, + workflow_node_execution_repository=node_repository, + restore_file_access=partial( + HumanInputSubmissionService(session_factory=sqlite_session_factory).grant_submitted_file_access, + app_id=app.id, + workflow_run_id="run", + ), + ) + + assert len(completed) == 1 + assert completed[0].outputs["result"] == submitted.value + @staticmethod def test_worker_executes_and_persists_a_workflow_through_the_injected_driver( app: Flask, sqlite_session: Session, sqlite_session_factory: sessionmaker[Session] diff --git a/api/tests/unit_tests/core/human_input_v2/test_im_sync_reconciliation.py b/api/tests/unit_tests/core/human_input_v2/test_im_sync_reconciliation.py index d0747d8d3a4..984c70df3d7 100644 --- a/api/tests/unit_tests/core/human_input_v2/test_im_sync_reconciliation.py +++ b/api/tests/unit_tests/core/human_input_v2/test_im_sync_reconciliation.py @@ -28,6 +28,7 @@ from core.human_input_v2.im_integration import ( SyncReconciler, ) from core.human_input_v2.im_integration.adapters import DirectoryEntry, ProviderUserId +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator from core.human_input_v2.shared import ( AccountId, ContactId, @@ -114,7 +115,7 @@ def test_sync_run_captures_complete_channel_revision() -> None: sync_run_id=IMSyncRunId("run-1"), channel_revision=_RUN.channel_revision, provider=IMProvider.FEISHU, - started_by_account_id=AccountId("account-1"), + started_by=AccountIMSyncInitiator(AccountId("account-1")), now=_NOW, ) diff --git a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_im_sync_queries.py b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_im_sync_queries.py index 8ba8fe1ad4f..4174b1704fb 100644 --- a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_im_sync_queries.py +++ b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_im_sync_queries.py @@ -11,6 +11,7 @@ from sqlalchemy.orm import Session from core.human_input_v2.entities import IMIdentityBindingStatus, IMProvider, IMSyncResultType from core.human_input_v2.im_integration import IMChannelRevision, IMSyncRun, SyncResultFact +from core.human_input_v2.im_integration.sync_records import SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import ContactId, IMIdentityId, IMSyncResultId, IMSyncRunId from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials from repositories.human_input_v2.im_binding_repository import IMBindingAssignment @@ -61,7 +62,7 @@ def _run(run_id: IMSyncRunId, created_at: datetime) -> IMSyncRun: sync_run_id=run_id, channel_revision=IMChannelRevision(str(_CHANNEL_ID), 1), provider=IMProvider.FEISHU, - started_by_account_id=None, + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), now=created_at, ) diff --git a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_mappers.py b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_mappers.py index 25a4f2b0dae..2a51ef6bdff 100644 --- a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_mappers.py +++ b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_mappers.py @@ -1,5 +1,6 @@ """Bidirectional mapping tests for every IM Control Plane record.""" +from dataclasses import replace from datetime import datetime import pytest @@ -23,6 +24,12 @@ from core.human_input_v2.im_integration import ( SyncResultFact, ) from core.human_input_v2.im_integration.adapters import ProviderUserId +from core.human_input_v2.im_integration.sync_records import ( + AccountIMSyncInitiator, + IMSyncInitiator, + SysUserId, + SysUserIMSyncInitiator, +) from core.human_input_v2.shared import ( AccountId, ContactId, @@ -52,7 +59,7 @@ def _run() -> IMSyncRun: sync_run_id=IMSyncRunId("run-1"), channel_revision=IMChannelRevision(str(_CHANNEL_ID), 1), provider=IMProvider.FEISHU, - started_by_account_id=AccountId("account-1"), + started_by=AccountIMSyncInitiator(AccountId("account-1")), now=_NOW, ) @@ -94,12 +101,30 @@ def _result() -> SyncResultFact: ) -def test_sync_run_mapping_round_trips_captured_revision_and_counts() -> None: - run = _run() +@pytest.mark.parametrize( + "started_by", + [AccountIMSyncInitiator(AccountId("account-1")), SysUserIMSyncInitiator(SysUserId("sys-user-1"))], +) +def test_sync_run_mapping_round_trips_captured_revision_counts_and_initiator(started_by: IMSyncInitiator) -> None: + run = replace(_run(), started_by=started_by) assert sync_run_from_record(sync_run_to_record(run)) == run +@pytest.mark.parametrize( + ("account_id", "sys_user_id"), + [(None, None), ("account-1", "sys-user-1")], + ids=["missing-initiator", "multiple-initiators"], +) +def test_sync_run_mapping_rejects_invalid_initiator_columns(account_id: str | None, sys_user_id: str | None) -> None: + record = sync_run_to_record(_run()) + record.started_by_account_id = account_id + record.started_by_sys_user_id = sys_user_id + + with pytest.raises(AssertionError): + sync_run_from_record(record) + + def test_sync_result_mapping_round_trips_all_structured_snapshots() -> None: result = _result() diff --git a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_repository.py b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_repository.py index 82ada3ffc85..f0fae5bbd07 100644 --- a/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_repository.py +++ b/api/tests/unit_tests/repositories/human_input_v2/im_integration/test_repository.py @@ -4,13 +4,16 @@ from __future__ import annotations from datetime import datetime +import pytest from sqlalchemy import Engine, func, select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from core.human_input_v2.entities import IMProvider, IMSyncResultType from core.human_input_v2.im_integration import IMChannelRevision, IMSyncRun, SyncResultFact +from core.human_input_v2.im_integration.sync_records import SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import IMSyncResultId, IMSyncRunId -from models.human_input_v2 import HumanInputIMSyncResult +from models.human_input_v2 import HumanInputIMSyncResult, HumanInputIMSyncRun from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials from repositories.human_input_v2.im_channel_repository import ( IMChannel, @@ -44,7 +47,7 @@ def _run() -> IMSyncRun: sync_run_id=IMSyncRunId("00000000-0000-0000-0000-000000000201"), channel_revision=IMChannelRevision(str(_CHANNEL_ID), 1), provider=IMProvider.FEISHU, - started_by_account_id=None, + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), now=_NOW, ) @@ -88,3 +91,29 @@ def test_diagnostic_append_uses_the_supplied_session_transaction(sqlite_engine: with Session(sqlite_engine) as session: assert session.scalar(select(func.count(HumanInputIMSyncResult.id))) == 0 + + +@pytest.mark.parametrize( + ("account_id", "sys_user_id"), + [ + (None, None), + ("00000000-0000-0000-0000-000000000501", "00000000-0000-0000-0000-000000000601"), + ], + ids=["missing-initiator", "multiple-initiators"], +) +def test_sync_run_requires_exactly_one_initiator( + sqlite_engine: Engine, account_id: str | None, sys_user_id: str | None +) -> None: + with Session(sqlite_engine) as session: + session.add( + HumanInputIMSyncRun( + channel_id=str(_CHANNEL_ID), + channel_config_version=1, + provider=IMProvider.FEISHU, + status=_run().status, + started_by_account_id=account_id, + started_by_sys_user_id=sys_user_id, + ) + ) + with pytest.raises(IntegrityError, match="initiator_exclusive"): + session.flush() diff --git a/api/tests/unit_tests/services/human_input_v2/test_im_channel_reconciliation_service.py b/api/tests/unit_tests/services/human_input_v2/test_im_channel_reconciliation_service.py index 359567aaad6..20df02ffe39 100644 --- a/api/tests/unit_tests/services/human_input_v2/test_im_channel_reconciliation_service.py +++ b/api/tests/unit_tests/services/human_input_v2/test_im_channel_reconciliation_service.py @@ -28,6 +28,7 @@ from core.human_input_v2.im_integration.adapters import ( ProviderUserId, ) from core.human_input_v2.im_integration.adapters.entities import Avatar +from core.human_input_v2.im_integration.sync_records import AccountIMSyncInitiator from core.human_input_v2.shared import AccountId, ContactId, IMSyncRunId, TenantId from extensions.ext_storage import storage from models.account import Account, TenantAccountJoin, TenantAccountRole @@ -187,7 +188,7 @@ def reconciliation_context(sqlite_engine: Engine) -> _ReconciliationContext: WorkspaceIMChannelWriter(session, _TENANT_ID, _ACCOUNT_ID).create(channel) decision = SQLAlchemyIMControlPlaneRepository(session, channel).create_or_get_active_run( sync_run_id=_RUN_ID, - started_by_account_id=_ACCOUNT_ID, + started_by=AccountIMSyncInitiator(_ACCOUNT_ID), now=_NOW, ) assert decision.run is not None @@ -214,17 +215,17 @@ def _create_active_run( with sessions.begin() as session: decision = SQLAlchemyIMControlPlaneRepository(session, channel).create_or_get_active_run( sync_run_id=sync_run_id, - started_by_account_id=_ACCOUNT_ID, + started_by=AccountIMSyncInitiator(_ACCOUNT_ID), now=_NOW, ) assert decision.run is not None -@pytest.mark.parametrize("has_initiator", [True, False]) -def test_workspace_sync_stores_avatar_with_initiator_or_owner( +@pytest.mark.parametrize("has_account_initiator", [True, False]) +def test_workspace_sync_stores_avatar_with_account_initiator_or_owner( reconciliation_context: _ReconciliationContext, mocker: MockerFixture, - has_initiator: bool, + has_account_initiator: bool, ) -> None: save = mocker.patch.object(storage, "save") with reconciliation_context.sessions.begin() as session: @@ -241,9 +242,11 @@ def test_workspace_sync_stores_avatar_with_initiator_or_owner( ), ] ) - if not has_initiator: - session.get_one(HumanInputIMSyncRun, str(_RUN_ID)).started_by_account_id = None - expected_creator = str(_ACCOUNT_ID) if has_initiator else owner.id + if not has_account_initiator: + run_record = session.get_one(HumanInputIMSyncRun, str(_RUN_ID)) + run_record.started_by_account_id = None + run_record.started_by_sys_user_id = "00000000-0000-0000-0000-000000000999" + expected_creator = str(_ACCOUNT_ID) if has_account_initiator else owner.id entry = replace(_successful_directory().entries[0], avatar=Avatar(mime_type="image/png", data=b"first avatar")) service = reconciliation_context.service(_Adapter(Directory((entry,)))) @@ -295,7 +298,9 @@ def test_missing_workspace_owner_fails_before_storing_avatars( ) -> None: save = mocker.patch.object(storage, "save") with reconciliation_context.sessions.begin() as session: - session.get_one(HumanInputIMSyncRun, str(_RUN_ID)).started_by_account_id = None + run_record = session.get_one(HumanInputIMSyncRun, str(_RUN_ID)) + run_record.started_by_account_id = None + run_record.started_by_sys_user_id = "00000000-0000-0000-0000-000000000999" entry = replace(_successful_directory().entries[0], avatar=Avatar(mime_type="image/png", data=b"avatar")) run = reconciliation_context.service(_Adapter(Directory((entry,)))).reconcile(_RUN_ID) @@ -726,7 +731,7 @@ def test_trigger_retires_abandoned_old_revision_and_creates_current_run( with reconciliation_context.sessions.begin() as session: repository = SQLAlchemyIMControlPlaneRepository(session, current_channel) decision = repository.create_or_get_active_run( - sync_run_id=_SECOND_RUN_ID, started_by_account_id=_ACCOUNT_ID, now=_NOW + sync_run_id=_SECOND_RUN_ID, started_by=AccountIMSyncInitiator(_ACCOUNT_ID), now=_NOW ) assert decision.run is not None assert decision.run.id == _SECOND_RUN_ID diff --git a/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_composition.py b/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_composition.py index fed4a3dfe8f..95c90d7b665 100644 --- a/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_composition.py +++ b/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_composition.py @@ -27,6 +27,7 @@ from core.human_input_v2.im_integration.adapters import ( SlackCredentials, ) from core.human_input_v2.im_integration.adapters.credentials import FeishuCredentials, IMProviderCredentials +from core.human_input_v2.im_integration.sync_records import SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import AccountId, ContactId, DeploymentScope, IMSyncRunId, TenantId, WorkspaceScope from libs.uuid_utils import uuidv7 from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials @@ -225,7 +226,7 @@ def _seed_run( DeploymentIMChannelWriter(session).create(channel) decision = SQLAlchemyIMControlPlaneRepository(session, channel).create_or_get_active_run( sync_run_id=_RUN_ID, - started_by_account_id=None, + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), now=_NOW, ) assert decision.run is not None diff --git a/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_worker.py b/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_worker.py index 2c93dbaad2b..efc697cb1f9 100644 --- a/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_worker.py +++ b/api/tests/unit_tests/services/human_input_v2/test_im_contact_sync_worker.py @@ -10,6 +10,7 @@ import pytest from core.human_input_v2.entities import IMProvider, IMSyncRunStatus from core.human_input_v2.im_integration import IMChannelRevision, IMSyncRun +from core.human_input_v2.im_integration.sync_records import SysUserId, SysUserIMSyncInitiator from core.human_input_v2.shared import IMSyncRunId, TenantId, WorkspaceScope from services.human_input_v2.im_contact_sync.coordinator import IMChannelReconciliationService, IMSyncRetryableError from services.human_input_v2.im_contact_sync.service import IMSyncService @@ -25,7 +26,7 @@ def _queued_run() -> IMSyncRun: sync_run_id=IMSyncRunId("run-1"), channel_revision=IMChannelRevision("channel-1", 1), provider=IMProvider.FEISHU, - started_by_account_id=None, + started_by=SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000999")), now=_NOW, ) diff --git a/api/tests/unit_tests/services/human_input_v2/test_im_sync_service.py b/api/tests/unit_tests/services/human_input_v2/test_im_sync_service.py index 7d962a2c26c..784c246b5f7 100644 --- a/api/tests/unit_tests/services/human_input_v2/test_im_sync_service.py +++ b/api/tests/unit_tests/services/human_input_v2/test_im_sync_service.py @@ -13,9 +13,16 @@ from sqlalchemy.orm import Session, sessionmaker from core.human_input_v2.entities import IMProvider, IMSyncResultType, IMSyncRunStatus from core.human_input_v2.im_integration import SyncResultFact +from core.human_input_v2.im_integration.sync_records import ( + AccountIMSyncInitiator, + IMSyncInitiator, + SysUserId, + SysUserIMSyncInitiator, +) from core.human_input_v2.shared import ( AccountId, ContactId, + DeploymentScope, DirectoryScope, IMBindingId, IMIdentityId, @@ -24,7 +31,9 @@ from core.human_input_v2.shared import ( TenantId, WorkspaceScope, ) +from models.account import Account from models.human_input_v2 import HumanInputIMSyncRun +from repositories.account.repository import SQLAlchemyAccountRepository from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials from repositories.human_input_v2.im_binding_repository import IMBindingAssignment from repositories.human_input_v2.im_channel_repository import ( @@ -37,6 +46,7 @@ from repositories.human_input_v2.im_identity_repository import IMIdentityObserva from repositories.human_input_v2.im_integration.mappers import sync_result_to_record from repositories.human_input_v2.sqlalchemy_im_binding_repository import SQLAlchemyIMBindingRepository from repositories.human_input_v2.sqlalchemy_im_channel_repository import ( + DeploymentIMChannelWriter, WorkspaceIMChannelReader, WorkspaceIMChannelWriter, ) @@ -90,6 +100,7 @@ class _ServiceContext: self.sessions, resolve_channel, lambda run_id, scope: self.dispatched.append((run_id, scope)), + account_query=SQLAlchemyAccountRepository(self.sessions), clock=lambda: _NOW, run_id_factory=lambda: _RUN_ID, ) @@ -108,17 +119,17 @@ def test_created_or_existing_active_run_is_serialized_and_dispatched_by_state( ) -> None: service = service_context.service() - created = service.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) - retried = service.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) + created = service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) + retried = service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) assert created == retried - assert created.channel_revision.channel_id == str(_CHANNEL_ID) + assert created.run.channel_revision.channel_id == str(_CHANNEL_ID) assert service_context.dispatched == [(_RUN_ID, _SCOPE), (_RUN_ID, _SCOPE)] with service_context.sessions.begin() as session: session.get_one(HumanInputIMSyncRun, str(_RUN_ID)).status = IMSyncRunStatus.RUNNING - running = service.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) - assert running.status is IMSyncRunStatus.RUNNING + running = service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) + assert running.run.status is IMSyncRunStatus.RUNNING assert service_context.dispatched == [(_RUN_ID, _SCOPE), (_RUN_ID, _SCOPE)] with service_context.sessions() as session: assert session.scalar(sa.select(sa.func.count(HumanInputIMSyncRun.id))) == 1 @@ -139,12 +150,13 @@ def test_dispatch_failure_occurs_after_queued_run_is_committed( service_context.sessions, resolve_channel, fail_dispatch, + account_query=SQLAlchemyAccountRepository(service_context.sessions), clock=lambda: _NOW, run_id_factory=lambda: _RUN_ID, ) with caplog.at_level(logging.ERROR), pytest.raises(IMSyncDispatchUnavailableError) as error_info: - service.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) + service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) assert isinstance(error_info.value.__cause__, ConnectionError) assert str(_RUN_ID) in caplog.text @@ -153,14 +165,19 @@ def test_dispatch_failure_occurs_after_queued_run_is_committed( def test_missing_or_stale_channel_fails_before_run_creation(service_context: _ServiceContext) -> None: - missing = IMSyncService(service_context.sessions, lambda _session, _scope: None, lambda _id, _scope: None) + missing = IMSyncService( + service_context.sessions, + lambda _session, _scope: None, + lambda _id, _scope: None, + account_query=SQLAlchemyAccountRepository(service_context.sessions), + ) with pytest.raises(IMChannelNotConfiguredError): - missing.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) + missing.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) with pytest.raises(IMSyncRevisionChangedError): service_context.service(channel_override=replace(_channel(), config_version=2)).create_or_get_active_run( _SCOPE, - _ACCOUNT_ID, + AccountIMSyncInitiator(_ACCOUNT_ID), ) with service_context.sessions() as session: @@ -169,7 +186,8 @@ def test_missing_or_stale_channel_fails_before_run_creation(service_context: _Se def test_latest_result_and_identity_queries_use_current_channel(service_context: _ServiceContext) -> None: service = service_context.service() - run = service.create_or_get_active_run(_SCOPE, _ACCOUNT_ID) + summary = service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) + run = summary.run with service_context.sessions.begin() as session: identities = SQLAlchemyIMIdentityRepository(session, _CHANNEL_ID) identities.create( @@ -218,7 +236,7 @@ def test_latest_result_and_identity_queries_use_current_channel(service_context: ) ) - assert service.get_latest_run(_SCOPE) == run + assert service.get_latest_run(_SCOPE) == summary result_page = service.list_latest_results(_SCOPE, IMSyncResultType.NOT_MATCHED, page=1, limit=20) identity_page = service.search_identities(_SCOPE, keyword="REVIEWER", page=1, limit=20) assert [result.id for result in result_page.items] == [IMSyncResultId("00000000-0000-0000-0000-000000000701")] @@ -229,3 +247,75 @@ def test_latest_result_and_identity_queries_use_current_channel(service_context: def test_latest_queries_fail_closed_without_a_current_run(service_context: _ServiceContext) -> None: with pytest.raises(IMSyncRunNotFoundError): service_context.service().get_latest_run(_SCOPE) + + +@pytest.mark.parametrize( + ("owner_scope", "started_by", "account_id", "sys_user_id", "display_name"), + [ + (_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID), _ACCOUNT_ID, None, "Workspace Owner"), + (DeploymentScope(), SysUserIMSyncInitiator(SysUserId(str(_ACCOUNT_ID))), None, str(_ACCOUNT_ID), ""), + ], + ids=["ce-saas-account", "ee-administrator"], +) +def test_sync_run_preserves_initiating_identity_when_another_caller_reuses_it( + sqlite_engine: Engine, + owner_scope: DirectoryScope, + started_by: IMSyncInitiator, + account_id: AccountId | None, + sys_user_id: str | None, + display_name: str, +) -> None: + sessions = sessionmaker(bind=sqlite_engine, expire_on_commit=False) + channel = _channel() + with sessions.begin() as session: + account = Account(name="Workspace Owner", email="owner@example.com") + account.id = str(_ACCOUNT_ID) + session.add(account) + if isinstance(owner_scope, WorkspaceScope): + WorkspaceIMChannelWriter(session, owner_scope.id, _ACCOUNT_ID).create(channel) + else: + DeploymentIMChannelWriter(session).create(channel) + service = IMSyncService( + sessions, + lambda _session, _scope: channel, + lambda _run_id, _scope: None, + account_query=SQLAlchemyAccountRepository(sessions), + ) + + created = service.create_or_get_active_run(owner_scope, started_by) + next_initiator = ( + AccountIMSyncInitiator(AccountId("00000000-0000-0000-0000-000000000202")) + if isinstance(started_by, AccountIMSyncInitiator) + else SysUserIMSyncInitiator(SysUserId("00000000-0000-0000-0000-000000000202")) + ) + reused = service.create_or_get_active_run(owner_scope, next_initiator) + + assert reused == created + assert created.initiator_display_name == display_name + assert service.get_latest_run(owner_scope).initiator_display_name == display_name + loaded = service.load_run(created.run.id) + assert loaded is not None + assert loaded.started_by == started_by + with sessions() as session: + stored = session.get_one(HumanInputIMSyncRun, str(created.run.id)) + assert stored.started_by_account_id == account_id + assert stored.started_by_sys_user_id == sys_user_id + + +def test_sync_run_returns_current_account_name(service_context: _ServiceContext) -> None: + with service_context.sessions.begin() as session: + account = Account(name="Original Admin", email="initiator@example.com") + account.id = str(_ACCOUNT_ID) + session.add(account) + service = service_context.service() + service.create_or_get_active_run(_SCOPE, AccountIMSyncInitiator(_ACCOUNT_ID)) + with service_context.sessions.begin() as session: + account = session.get_one(Account, str(_ACCOUNT_ID)) + account.name = "Renamed Admin" + + assert service.get_latest_run(_SCOPE).initiator_display_name == "Renamed Admin" + + with service_context.sessions.begin() as session: + session.delete(session.get_one(Account, str(_ACCOUNT_ID))) + + assert service.get_latest_run(_SCOPE).initiator_display_name == "" diff --git a/api/tests/unit_tests/services/human_input_v2/test_submission_service.py b/api/tests/unit_tests/services/human_input_v2/test_submission_service.py index d53db53a23d..46015d3bda0 100644 --- a/api/tests/unit_tests/services/human_input_v2/test_submission_service.py +++ b/api/tests/unit_tests/services/human_input_v2/test_submission_service.py @@ -1,17 +1,31 @@ """Accepted submissions retain the exact authenticated delivery across all surfaces.""" from collections.abc import Callable +from contextlib import nullcontext +from datetime import timedelta from hashlib import sha256 +from uuid import uuid4 import pytest from flask.testing import FlaskClient from sqlalchemy import event -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom +from core.app.file_access.controller import DatabaseFileAccessController +from core.app.file_access.scope import FileAccessScope, bind_file_access_scope +from core.app.workflow.file_runtime import DifyWorkflowFileRuntime from core.human_input_v2.entities import IMProvider +from core.human_input_v2.resolved_form import ResolvedForm +from core.human_input_v2.shared.values import RecipientId, TenantId +from core.workflow.file_reference import build_file_reference from core.workflow.nodes.human_input.enums import HumanInputFormStatus +from graphon.file import File, FileTransferMethod, FileType +from graphon.variables.segments import ArrayFileSegment, FileSegment, StringSegment +from libs.datetime_utils import naive_utc_now from models.human_input_v2 import HumanInputForm from repositories.human_input_v2.delivery_repository import DeliveryCreateParams +from repositories.human_input_v2.form_repository import Form, FormCreateParams from repositories.human_input_v2.im_channel_repository import IMChannelId from repositories.human_input_v2.sqlalchemy_delivery_repository import SQLAlchemyDeliveryRepository from repositories.human_input_v2.sqlalchemy_form_repository import SQLAlchemyFormRepository @@ -24,6 +38,7 @@ from services.human_input_v2.submission_service import ( ServiceAPIFormActor, WebAppFormActor, ) +from tests.helpers.unit.model_factories import make_upload_file from tests.unit_tests.services.human_input_v2.test_form_delivery_runtime import Context, PublicFormApproval pytest_plugins = ( @@ -32,6 +47,112 @@ pytest_plugins = ( ) +@pytest.mark.parametrize("transfer_method", [FileTransferMethod.LOCAL_FILE, FileTransferMethod.REMOTE_URL]) +@pytest.mark.parametrize( + ("form_tenant", "form_app", "form_run", "status", "accessible"), + [ + ("tenant", "app", "run", HumanInputFormStatus.SUBMITTED, True), + ("other-tenant", "app", "run", HumanInputFormStatus.SUBMITTED, False), + ("tenant", "other-app", "run", HumanInputFormStatus.SUBMITTED, False), + ("tenant", "app", "other-run", HumanInputFormStatus.SUBMITTED, False), + ("tenant", "app", "run", HumanInputFormStatus.WAITING, False), + ("tenant", "app", "run", HumanInputFormStatus.TIMEOUT, False), + ("tenant", "app", "run", HumanInputFormStatus.EXPIRED, False), + ], +) +def test_restored_file_access_requires_an_accepted_submission_in_the_current_run( + sqlite_session_factory: sessionmaker[Session], + monkeypatch: pytest.MonkeyPatch, + transfer_method: FileTransferMethod, + form_tenant: str, + form_app: str, + form_run: str, + status: HumanInputFormStatus, + accessible: bool, +) -> None: + uploaded = make_upload_file(tenant_id="tenant", created_by="approver") + list_uploaded = make_upload_file(tenant_id="tenant", created_by="approver") + unrelated = make_upload_file(tenant_id="tenant", created_by="approver") + files = [ + File( + file_type=FileType.DOCUMENT, + transfer_method=transfer_method, + reference=build_file_reference(record_id=record.id), + remote_url="https://example.com/document.txt" if transfer_method == FileTransferMethod.REMOTE_URL else None, + filename=record.name, + extension=".txt", + mime_type=record.mime_type, + size=record.size, + ) + for record in (uploaded, list_uploaded) + ] + now = naive_utc_now() + with sqlite_session_factory.begin() as session: + session.add_all([uploaded, list_uploaded, unrelated]) + repository = SQLAlchemyFormRepository(session, TenantId(form_tenant), form_app) + form = repository.create_form( + FormCreateParams( + workflow_run_id=form_run, + node_execution_id=str(uuid4()), + resolved_form=ResolvedForm(title="Approval", parts=(), actions=(), legacy_form_content=""), + expiration_time=now + timedelta(hours=1), + global_timeout_deadline=now + timedelta(days=1), + ) + ) + if status == HumanInputFormStatus.SUBMITTED: + accepted = repository.submit_form( + form.id, + submitted_by_recipient_id=RecipientId("recipient"), + submitted_by_delivery_id="delivery", + selected_action_id="approve", + raw_submission_inputs={"untrusted_file": unrelated.id}, + normalized_submission_data={ + "single": FileSegment(value=files[0]), + "multiple": ArrayFileSegment(value=[files[1]]), + "empty": ArrayFileSegment(value=[]), + "text": StringSegment(value=unrelated.id), + }, + ) + assert isinstance(accepted, Form) + else: + session.get_one(HumanInputForm, form.id).status = status + + runtime = DifyWorkflowFileRuntime(file_access_controller=DatabaseFileAccessController()) + monkeypatch.setattr(runtime, "_secret_key", lambda: b"test-key") + scope = FileAccessScope( + tenant_id="tenant", user_id="end-user", user_from=UserFrom.END_USER, invoke_from=InvokeFrom.WEB_APP + ) + service = HumanInputSubmissionService(session_factory=sqlite_session_factory) + with bind_file_access_scope(scope): + service.grant_submitted_file_access(app_id="app", workflow_run_id="run") + for record in (uploaded, list_uploaded): + if accessible: + assert runtime.resolve_upload_file_uri(upload_file_id=record.id).startswith( + f"/files/{record.id}/file-preview?" + ) + else: + with pytest.raises(ValueError, match="not found"): + runtime.resolve_upload_file_uri(upload_file_id=record.id) + with pytest.raises(ValueError, match="not found"): + runtime.resolve_upload_file_uri(upload_file_id=unrelated.id) + + with bind_file_access_scope(scope), pytest.raises(ValueError, match="not found"): + runtime.resolve_upload_file_uri(upload_file_id=uploaded.id) + + +@pytest.mark.parametrize("account_scope", [False, True], ids=["no-scope", "account"]) +def test_restoring_file_access_outside_end_user_scope_does_not_read_storage( + unbound_session_factory: sessionmaker[Session], account_scope: bool +) -> None: + scope = FileAccessScope( + tenant_id="tenant", user_id="account", user_from=UserFrom.ACCOUNT, invoke_from=InvokeFrom.DEBUGGER + ) + with bind_file_access_scope(scope) if account_scope else nullcontext(): + HumanInputSubmissionService(session_factory=unbound_session_factory).grant_submitted_file_access( + app_id="app", workflow_run_id="run" + ) + + @pytest.fixture(params=[False, True], ids=["commit", "rollback-then-retry"]) def fail_commit(request: pytest.FixtureRequest) -> bool: return request.param is True diff --git a/oxlint-suppressions.json b/oxlint-suppressions.json index b9524889951..c9ce9bab601 100644 --- a/oxlint-suppressions.json +++ b/oxlint-suppressions.json @@ -1562,11 +1562,6 @@ "count": 2 } }, - "web/app/components/base/prompt-editor/plugins/hitl-input-block/component-ui.tsx": { - "better-tailwindcss/enforce-logical-properties": { - "count": 4 - } - }, "web/app/components/base/prompt-editor/plugins/hitl-input-block/index.tsx": { "no-barrel-files/no-barrel-files": { "count": 3 @@ -1575,11 +1570,6 @@ "count": 2 } }, - "web/app/components/base/prompt-editor/plugins/hitl-input-block/input-field.tsx": { - "better-tailwindcss/enforce-logical-properties": { - "count": 1 - } - }, "web/app/components/base/prompt-editor/plugins/hitl-input-block/pre-populate.tsx": { "better-tailwindcss/enforce-logical-properties": { "count": 2 diff --git a/packages/contracts/generated/api/console/workspaces/orpc.gen.ts b/packages/contracts/generated/api/console/workspaces/orpc.gen.ts index 3827461929a..543829cbb3d 100644 --- a/packages/contracts/generated/api/console/workspaces/orpc.gen.ts +++ b/packages/contracts/generated/api/console/workspaces/orpc.gen.ts @@ -1785,12 +1785,12 @@ export const results = { } /** - * Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; the response does not include started_by. + * Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; initiator_display_name is resolved from the initiating Account. */ export const get23 = oc .route({ description: - 'Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; the response does not include started_by.', + 'Return the latest IM sync run summary. The UI uses finished_at as the explicit sync time; initiator_display_name is resolved from the initiating Account.', inputStructure: 'detailed', method: 'GET', operationId: 'getWorkspacesCurrentHumanInputV2ImSyncRunsLatest', @@ -1804,12 +1804,16 @@ export const latest = { results, } +/** + * Start or reuse a workspace run using the authenticated Account ID + */ export const post22 = oc .route({ inputStructure: 'detailed', method: 'POST', operationId: 'postWorkspacesCurrentHumanInputV2ImSyncRuns', path: '/workspaces/current/human-input/v2/im-sync-runs', + summary: 'Start or reuse a workspace run using the authenticated Account ID', tags: ['console'], }) .output(zPostWorkspacesCurrentHumanInputV2ImSyncRunsResponse) diff --git a/packages/contracts/generated/api/console/workspaces/types.gen.ts b/packages/contracts/generated/api/console/workspaces/types.gen.ts index 5d837ff28b1..ef083dc4f83 100644 --- a/packages/contracts/generated/api/console/workspaces/types.gen.ts +++ b/packages/contracts/generated/api/console/workspaces/types.gen.ts @@ -2034,6 +2034,7 @@ export type HumanInputContact = { export type HumanInputContactSummary = { avatar_url?: string created_at: number + email?: string | null id: string name: string } @@ -2060,6 +2061,7 @@ export type ImSyncRun = { error_message?: string | null finished_at?: number | null id: string + initiator_display_name: string provider: ImProvider result_counts: ImSyncRunResultCounts started_at?: number | null diff --git a/packages/contracts/generated/api/console/workspaces/zod.gen.ts b/packages/contracts/generated/api/console/workspaces/zod.gen.ts index 12fb6f2239d..4935335566f 100644 --- a/packages/contracts/generated/api/console/workspaces/zod.gen.ts +++ b/packages/contracts/generated/api/console/workspaces/zod.gen.ts @@ -1752,6 +1752,7 @@ export const zUnchangedInput = z.object({ export const zHumanInputContactSummary = z.object({ avatar_url: z.string().optional().default(''), created_at: z.int(), + email: z.string().nullish(), id: z.string(), name: z.string(), }) @@ -3393,8 +3394,8 @@ export const zImSyncRunStatus = z.enum(['failed', 'queued', 'running', 'succeede * * One IM sync run snapshot. * - * The latest-only UI displays ``finished_at`` as the explicit sync time. The - * transport contract intentionally does not expose a ``started_by`` actor. + * The latest-only UI displays ``finished_at`` as the explicit sync time. + * The initiator display name is resolved from the initiating Dify Account. */ export const zImSyncRun = z.object({ channel_config_version: z.int().gte(1), @@ -3402,6 +3403,7 @@ export const zImSyncRun = z.object({ error_message: z.string().nullish(), finished_at: z.int().nullish(), id: z.string(), + initiator_display_name: z.string(), provider: zImProvider, result_counts: zImSyncRunResultCounts, started_at: z.int().nullish(), diff --git a/web/app/components/base/prompt-editor/index.tsx b/web/app/components/base/prompt-editor/index.tsx index 7d3919e664d..04d3a63c58e 100644 --- a/web/app/components/base/prompt-editor/index.tsx +++ b/web/app/components/base/prompt-editor/index.tsx @@ -97,9 +97,12 @@ type PromptEditorAriaProps = Pick< | 'aria-labelledby' | 'aria-describedby' | 'aria-invalid' + | 'aria-multiline' + | 'aria-disabled' > type PromptEditorProps = PromptEditorAriaProps & { + id?: string instanceId?: string children?: React.ReactNode compact?: boolean @@ -143,6 +146,9 @@ const PromptEditor: FC = ({ 'aria-labelledby': ariaLabelledBy, 'aria-describedby': ariaDescribedBy, 'aria-invalid': ariaInvalid, + 'aria-multiline': ariaMultiline, + 'aria-disabled': ariaDisabled, + id, instanceId, children, compact, @@ -244,12 +250,15 @@ const PromptEditor: FC = ({
{ - throw error - }, - nodes: [HITLInputNode], - }} - > - - , + const initialConfig = { + namespace: `hitl-input-test-${crypto.randomUUID()}`, + onError: (error: Error) => { + throw error + }, + nodes: [HITLInputNode], + } + const subject = (readonly = defaultProps.readonly) => ( + + + ) + const utils = render(subject()) return { ...utils, onChange, onRename, onRemove, + rerenderReadonly: (readonly: boolean) => utils.rerender(subject(readonly)), } } @@ -113,56 +113,41 @@ describe('HITLInputComponentUI', () => { ).not.toBeInTheDocument() }) - it('should close the edit modal when readonly becomes true', async () => { + it('should close the edit modal while readonly and allow reopening when editing resumes', async () => { const user = userEvent.setup() - let setReadonlyValue: ((readonly: boolean) => void) | undefined - const Harness = () => { - const [readonly, setReadonly] = useState(false) - const [namespace] = useState(() => `hitl-input-test-${crypto.randomUUID()}`) - - useEffect(() => { - setReadonlyValue = setReadonly - return () => { - setReadonlyValue = undefined - } - }, []) - - return ( - { - throw error - }, - nodes: [HITLInputNode], - }} - > - - - ) - } - - render() + const { rerenderReadonly } = renderComponent() await user.click(await screen.findByRole('button', { name: 'common.operation.edit' })) expect(await screen.findByRole('textbox')).toBeInTheDocument() - act(() => { - setReadonlyValue?.(true) - }) + rerenderReadonly(true) await waitFor(() => { expect(screen.queryByRole('textbox')).not.toBeInTheDocument() }) + + expect( + screen.queryByRole('button', { name: 'common.operation.edit' }), + ).not.toBeInTheDocument() + rerenderReadonly(false) + await user.click(screen.getByRole('button', { name: 'common.operation.edit' })) + + expect(await screen.findByRole('textbox')).toBeInTheDocument() + }) + + it('should enable edit and remove actions when an initially readonly field becomes editable', async () => { + const user = userEvent.setup() + const { onRemove, rerenderReadonly } = renderComponent({ readonly: true }) + + rerenderReadonly(false) + await user.click(screen.getByRole('button', { name: 'common.operation.edit' })) + expect(await screen.findByRole('textbox')).toBeInTheDocument() + await user.click(screen.getByRole('button', { name: 'common.operation.cancel' })) + await waitFor(() => expect(screen.queryByRole('textbox')).not.toBeInTheDocument()) + + await user.click(screen.getByRole('button', { name: 'common.operation.remove' })) + expect(onRemove).toHaveBeenCalledExactlyOnceWith(varName) }) it('should render select option summary for constant options', () => { @@ -224,6 +209,26 @@ describe('HITLInputComponentUI', () => { expect(onRemove).toHaveBeenCalledWith(varName) expect(onRemove).toHaveBeenCalledTimes(1) }) + + it('should remove exactly once per click after repeated readonly transitions', async () => { + const user = userEvent.setup() + const { onRemove, rerenderReadonly } = renderComponent() + + await user.click(screen.getByRole('button', { name: 'common.operation.remove' })) + expect(onRemove).toHaveBeenCalledExactlyOnceWith(varName) + + for (const expectedCalls of [2, 3]) { + rerenderReadonly(true) + expect( + screen.queryByRole('button', { name: 'common.operation.remove' }), + ).not.toBeInTheDocument() + rerenderReadonly(false) + await user.click(screen.getByRole('button', { name: 'common.operation.remove' })) + + expect(onRemove).toHaveBeenCalledTimes(expectedCalls) + expect(onRemove).toHaveBeenLastCalledWith(varName) + } + }) }) describe('Edit flow', () => { diff --git a/web/app/components/base/prompt-editor/plugins/hitl-input-block/__tests__/input-field.spec.tsx b/web/app/components/base/prompt-editor/plugins/hitl-input-block/__tests__/input-field.spec.tsx index a233f56520d..8b74dfddb24 100644 --- a/web/app/components/base/prompt-editor/plugins/hitl-input-block/__tests__/input-field.spec.tsx +++ b/web/app/components/base/prompt-editor/plugins/hitl-input-block/__tests__/input-field.spec.tsx @@ -164,6 +164,11 @@ describe('InputField', () => { await user.click( await screen.findByRole('option', { name: /appDebug\.variableConfig\.select\b/ }), ) + await user.click(screen.getByRole('button', { name: 'appDebug.variableConfig.addOption' })) + await user.type( + screen.getByRole('textbox', { name: 'appDebug.variableConfig.options 1' }), + 'alpha', + ) await user.click(screen.getByRole('button', { name: 'common.operation.save' })) expect(onChange).toHaveBeenCalledExactlyOnceWith( @@ -467,7 +472,7 @@ describe('InputField', () => { }) }) - it('should switch to select payload when field type changes', async () => { + it('should block insertion of a select field without options', async () => { const user = userEvent.setup() const onChange = vi.fn() @@ -495,21 +500,55 @@ describe('InputField', () => { }), ) - expect(onChange).toHaveBeenCalledTimes(1) - expect(onChange.mock.calls[0]![0]).toEqual({ - type: InputVarType.select, - output_variable_name: 'valid_name', - option_source: { - type: 'constant', - selector: [], - value: [], - }, - }) + const error = 'appDebug.variableConfig.errorMsg.atLeastOneOption' + expect(screen.getByRole('alert')).toHaveTextContent(error) expect( - screen.queryByText( - /workflowHumanInput\.nodes\.humanInput\.insertInputField\.prePopulateField/i, - ), - ).not.toBeInTheDocument() + screen.getByRole('button', { name: 'appDebug.variableConfig.addOption' }), + ).toHaveAccessibleDescription(error) + expect(onChange).not.toHaveBeenCalled() + }) + + it('should reject blank and duplicate select options until they are corrected', async () => { + const user = userEvent.setup() + const onChange = vi.fn() + render( + , + ) + + const option = screen.getByRole('textbox', { name: 'appDebug.variableConfig.options 2' }) + const save = screen.getByRole('button', { name: 'common.operation.save' }) + await user.click(save) + expect(screen.getByRole('alert')).toHaveTextContent('common.errorMsg.fieldRequired') + expect(option).toHaveAccessibleDescription(/common\.errorMsg\.fieldRequired/) + expect(onChange).not.toHaveBeenCalled() + + await user.clear(option) + await user.type(option, ' alpha ') + await user.keyboard('{Control>}{Enter}{/Control}') + expect(screen.getByRole('alert')).toHaveTextContent( + 'appDebug.variableConfig.errorMsg.optionRepeat', + ) + expect(onChange).not.toHaveBeenCalled() + + await user.clear(option) + await user.type(option, ' Alpha ') + await user.click(save) + expect(screen.queryByRole('alert')).not.toBeInTheDocument() + expect(onChange).toHaveBeenCalledExactlyOnceWith({ + type: InputVarType.select, + output_variable_name: 'choice', + option_source: { type: 'constant', selector: [], value: ['alpha', ' Alpha '] }, + }) }) it('should keep paragraph pre-populate editor available after switching back to paragraph', async () => { @@ -677,15 +716,16 @@ describe('InputField', () => { }) }) - it('should only allow array[string] variables for select option source', async () => { + it('should save a select using an array[string] variable without constant options', async () => { const user = userEvent.setup() + const onChange = vi.fn() render( , ) @@ -703,9 +743,17 @@ describe('InputField', () => { ) await user.click(screen.getByRole('button', { name: /common.setVarValuePlaceholder/ })) - expect(await screen.findByText('choices')).toBeInTheDocument() + const choices = await screen.findByText('choices') expect(screen.queryByText('text', { exact: true })).not.toBeInTheDocument() expect(screen.queryByText('files', { exact: true })).not.toBeInTheDocument() + await user.click(choices) + await user.click(screen.getByRole('button', { name: /insertInputField\.insert/ })) + + expect(onChange).toHaveBeenCalledExactlyOnceWith({ + type: InputVarType.select, + output_variable_name: 'valid_name', + option_source: { type: 'variable', selector: ['code', 'choices'], value: [] }, + }) }) it('should save single file upload settings', async () => { diff --git a/web/app/components/base/prompt-editor/plugins/hitl-input-block/component-ui.tsx b/web/app/components/base/prompt-editor/plugins/hitl-input-block/component-ui.tsx index 86b76af5f88..e36bfc4c5ae 100644 --- a/web/app/components/base/prompt-editor/plugins/hitl-input-block/component-ui.tsx +++ b/web/app/components/base/prompt-editor/plugins/hitl-input-block/component-ui.tsx @@ -72,7 +72,7 @@ const HITLInputComponentUI: FC = ({ return () => { if (editBtn) editBtn.removeEventListener('click', showEditModal) } - }, [showEditModal]) + }, [showEditModal, readonly]) const removeBtnRef = useRef(null) useEffect(() => { @@ -83,7 +83,7 @@ const HITLInputComponentUI: FC = ({ return () => { if (removeBtn) removeBtn.removeEventListener('click', removeHandler) } - }, [onRemove, varName]) + }, [onRemove, varName, readonly]) const handleChange = useCallback( (newPayload: FormInputItem) => { @@ -131,8 +131,8 @@ const HITLInputComponentUI: FC = ({ }, [inputTypeLabel, paragraphDefault?.value, resolvedFormInput, t]) return ( -
-
+
+
@@ -140,7 +140,7 @@ const HITLInputComponentUI: FC = ({
-
+
{variableSelector ? ( = ({ onChange, onCancel, }) => { - const rootRef = useRef(null) + const rootRef = useRef(null) const outputVariableNameInputId = React.useId() + const optionsErrorId = React.useId() const { t } = useTranslation(['appDebug', 'common', 'workflow', 'workflowHumanInput']) const [tempPayload, setTempPayload] = useState( () => payload || createDefaultParagraphFormInput(), @@ -100,8 +101,26 @@ const InputField: React.FC = ({ const nameValid = useMemo(() => { return !!tempPayload.output_variable_name.trim() && !variableNameError }, [tempPayload.output_variable_name, variableNameError]) + + let optionsError = '' + if (isSelectFormInput(tempPayload) && tempPayload.option_source.type === 'constant') { + const options = tempPayload.option_source.value.map((option) => option.trim()) + const blankOptionIndex = options.findIndex((option) => !option) + if (options.length === 0) { + optionsError = t(($) => $['variableConfig.errorMsg.atLeastOneOption'], { ns: 'appDebug' }) + } else if (blankOptionIndex !== -1) { + const label = t(($) => $['variableConfig.options'], { ns: 'appDebug' }) + optionsError = t(($) => $['errorMsg.fieldRequired'], { + ns: 'common', + field: `${label} ${blankOptionIndex + 1}`, + }) + } else if (new Set(options).size !== options.length) { + optionsError = t(($) => $['variableConfig.errorMsg.optionRepeat'], { ns: 'appDebug' }) + } + } + const handleSave = useCallback(() => { - if (!nameValid) return + if (!nameValid || optionsError) return if (isFileListFormInput(tempPayload)) { const value = tempPayload.number_limits ?? 5 if (!Number.isFinite(value)) { @@ -113,7 +132,7 @@ const InputField: React.FC = ({ } } onChange(tempPayload) - }, [nameValid, onChange, tempPayload]) + }, [nameValid, onChange, optionsError, tempPayload]) const handleTypeChange = useCallback((item: TypeSelectItem) => { setTempPayload((prev) => createDefaultFormInputByType(item.value as FormInputItem['type'], prev.output_variable_name), @@ -219,7 +238,7 @@ const InputField: React.FC = ({ event.preventDefault() event.stopPropagation() if (event.repeat) return - handleSave() + rootRef.current?.requestSubmit() }, { enabled: nameValid, @@ -230,9 +249,27 @@ const InputField: React.FC = ({ ) return ( -
{ + event.preventDefault() + event.stopPropagation() + handleSave() + }} + onKeyDown={(event) => { + // Preserve Mod+Enter as the save shortcut while editing a field. + if ( + event.key === 'Enter' && + !event.metaKey && + !event.ctrlKey && + !event.nativeEvent.isComposing && + event.target instanceof HTMLInputElement && + event.target.form === event.currentTarget + ) + event.preventDefault() + }} >
@@ -310,7 +347,7 @@ const InputField: React.FC = ({ filterVar={(varPayload) => varPayload.type === VarType.arrayString} /> @@ -324,7 +361,18 @@ const InputField: React.FC = ({ + {optionsError && ( + + )} = ({
- + {isEdit ? ( - ) : ( -
-
+ ) } diff --git a/web/app/components/base/prompt-editor/prompt-editor-content.tsx b/web/app/components/base/prompt-editor/prompt-editor-content.tsx index 55671824a1c..c00299656c0 100644 --- a/web/app/components/base/prompt-editor/prompt-editor-content.tsx +++ b/web/app/components/base/prompt-editor/prompt-editor-content.tsx @@ -65,9 +65,12 @@ type PromptEditorContentAriaProps = Pick< | 'aria-labelledby' | 'aria-describedby' | 'aria-invalid' + | 'aria-multiline' + | 'aria-disabled' > type PromptEditorContentProps = PromptEditorContentAriaProps & { + id?: string compact?: boolean className?: string placeholder?: string | React.ReactNode @@ -114,6 +117,9 @@ const PromptEditorContent: FC = ({ 'aria-labelledby': ariaLabelledBy, 'aria-describedby': ariaDescribedBy, 'aria-invalid': ariaInvalid, + 'aria-multiline': ariaMultiline, + 'aria-disabled': ariaDisabled, + id, compact, className, placeholder, @@ -147,12 +153,15 @@ const PromptEditorContent: FC = ({ subject.element().textContent).toBe('Converted approval') await page.getByRole('button', { name: 'common.operation.cancel' }).click() - await expect - .element(page.getByRole('textbox', { name: 'workflow.nodes.humanInputV2.template.subject' })) - .not.toBeInTheDocument() + await expect.element(subject).not.toBeInTheDocument() await lastRun.click() await expect.element(lastRun).toHaveAttribute('aria-selected', 'true') diff --git a/web/app/components/workflow/nodes/human-input-v2/__tests__/message-template-subject.browser.spec.tsx b/web/app/components/workflow/nodes/human-input-v2/__tests__/message-template-subject.browser.spec.tsx new file mode 100644 index 00000000000..8e695986ad1 --- /dev/null +++ b/web/app/components/workflow/nodes/human-input-v2/__tests__/message-template-subject.browser.spec.tsx @@ -0,0 +1,125 @@ +import { + Dialog, + DialogBody, + DialogContent, + DialogHeader, + DialogTitle, +} from '@langgenius/dify-ui/dialog' +import { Field, FieldLabel } from '@langgenius/dify-ui/field' +import { detectPlatform } from '@tanstack/react-hotkeys' +import { QueryClientProvider } from '@tanstack/react-query' +import { useState } from 'react' +import { ReactFlowProvider } from 'reactflow' +import { page, userEvent } from 'vite-plus/test/browser' +import { render } from 'vitest-browser-react' +import { createNode } from '@/app/components/workflow/__tests__/fixtures' +import { BlockEnum, VarType } from '@/app/components/workflow/types' +import { seedModelProviderQueries } from '@/test/console/model-provider' +import { createTestQueryClient } from '@/test/query-client' +import MessageTemplateSubject from '../components/message-template-subject' + +const insertVariableLabel = 'workflow.nodes.humanInputV2.template.insertVariable' +const searchVariableLabel = 'workflow.common.searchVar' + +function MessageTemplateSubjectHarness() { + const [subject, setSubject] = useState('Before placeholder After') + const [queryClient] = useState(() => { + const client = createTestQueryClient() + seedModelProviderQueries(client) + return client + }) + return ( + + + + + + Message template + + + + + Subject + + + + {subject} +