From 488a8d9307dbbb4d144c368c5fd8671996817bc8 Mon Sep 17 00:00:00 2001 From: twwu Date: Tue, 3 Nov 2026 16:16:39 +0800 Subject: [PATCH] fix(workflow): preserve final draft saves across editor teardown Dispatch keepalive snapshots before teardown and persist save receipts in the draft transaction. Validate base receipts and named predecessors to reject stale saves even when the graph returns to its original hash. Retain unacknowledged saves and variable patches for recovery, propagate receipts through refresh, import, and collaboration, and support UUID generation on HTTP deployments. Include the migration and API contracts. --- api/controllers/console/app/workflow.py | 62 +- ...b3f72457_add_workflow_draft_client_save.py | 30 + api/models/workflow.py | 3 + api/openapi/markdown/console-openapi.md | 13 + .../workflow_draft_sync_repository.py | 1 + .../workflow_collaboration_service.py | 46 +- api/services/workflow_draft_sync_service.py | 2 + api/services/workflow_service.py | 73 ++- .../test_workflow_draft_sync_repository.py | 13 +- .../test_workflow_draft_client_save.py | 489 +++++++++++++++ .../controllers/console/app/test_workflow.py | 51 +- ...b3f72457_add_workflow_draft_client_save.py | 63 ++ .../test_workflow_draft_sync_repository.py | 3 + .../test_workflow_collaboration_service.py | 107 +++- .../test_workflow_draft_sync_service.py | 7 + .../generated/api/console/apps/types.gen.ts | 10 + .../generated/api/console/apps/zod.gen.ts | 13 + .../generated/api/console/rag/types.gen.ts | 1 + .../generated/api/console/rag/zod.gen.ts | 1 + .../api/console/snippets/types.gen.ts | 1 + .../generated/api/console/snippets/zod.gen.ts | 1 + .../__tests__/workflow-main.creation.spec.tsx | 1 + .../__tests__/workflow-main.spec.tsx | 167 ++++++ .../workflow-app/components/workflow-main.tsx | 10 +- .../__tests__/use-nodes-sync-draft.spec.ts | 562 +++++++++++++++++- .../hooks/__tests__/use-workflow-init.spec.ts | 5 + ...use-workflow-refresh-draft.builder.spec.ts | 57 +- .../use-workflow-refresh-draft.spec.tsx | 8 + .../hooks/use-nodes-sync-draft.ts | 164 +++-- .../workflow-app/hooks/use-workflow-init.ts | 4 +- .../hooks/use-workflow-refresh-draft.ts | 8 +- .../__tests__/update-dsl-modal.spec.tsx | 8 +- .../__tests__/workflow-edge-events.spec.tsx | 6 + ...n-manager.socket-and-subscriptions.spec.ts | 74 ++- .../core/collaboration-manager.ts | 18 +- .../collaboration/core/server-draft-sync.ts | 8 +- .../collaboration/types/collaboration.ts | 8 + .../components/workflow/hooks-store/store.ts | 2 + web/app/components/workflow/index.tsx | 6 +- .../store/workflow/workflow-draft-slice.ts | 166 +++++- .../components/workflow/update-dsl-modal.tsx | 11 +- web/types/workflow.ts | 3 +- 42 files changed, 2159 insertions(+), 127 deletions(-) create mode 100644 api/migrations/versions/2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py create mode 100644 api/tests/e2e_tests/containers/services/test_workflow_draft_client_save.py create mode 100644 api/tests/unit_tests/migrations/versions/test_2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py diff --git a/api/controllers/console/app/workflow.py b/api/controllers/console/app/workflow.py index 1258204c463..6c742454fe1 100644 --- a/api/controllers/console/app/workflow.py +++ b/api/controllers/console/app/workflow.py @@ -3,6 +3,7 @@ import logging from collections.abc import Sequence from datetime import datetime from typing import Any, Literal, NotRequired, Self, TypedDict +from uuid import UUID from flask import abort, request from flask_restx import Resource @@ -105,7 +106,12 @@ from services.errors.llm import InvokeRateLimitError from services.errors.workflow_service import DuplicateDraftWorkflowError, EmptyWorkflowConfirmationRequiredError from services.workflow.webhook_policy import WorkflowWebhookLimitError from services.workflow_ref_service import WorkflowRefService -from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService +from services.workflow_service import ( + DraftSavePredecessor, + DraftWorkflowDeletionError, + WorkflowInUseError, + WorkflowService, +) from services.workflow_variable_reference_validator import advisory_variable_reference_warning logger = logging.getLogger(__name__) @@ -148,12 +154,27 @@ class SyncEnvironmentVariablePatchPayload(BaseModel): return self +class PredecessorDraftSavePayload(BaseModel): + model_config = ConfigDict(extra="forbid") + + client_save_id: UUID + environment_variable_patch: SyncEnvironmentVariablePatchPayload | None = None + + class SyncDraftWorkflowPayload(BaseModel): model_config = ConfigDict(extra="forbid") graph: dict[str, Any] features: dict[str, Any] hash: str | None = None + client_save_id: UUID | None = Field(default=None, description="Unique ID for this editor save attempt.") + base_client_save_id: UUID | None = Field( + default=None, description="Receipt ID returned with the editor's base draft." + ) + predecessor_saves: list[PredecessorDraftSavePayload] = Field( + default_factory=list, + description="Pending saves in editor order, including inherited patches; requires client_save_id.", + ) force: bool = Field(default=False, strict=True, description="Explicitly confirm saving an empty workflow graph.") is_collaborative: bool = Field(default=False, alias="_is_collaborative") environment_variable_patch: SyncEnvironmentVariablePatchPayload | None = None @@ -161,6 +182,16 @@ class SyncDraftWorkflowPayload(BaseModel): default_factory=list, ) + @model_validator(mode="after") + def validate_predecessor_saves(self) -> Self: + """Require a receipt for any save that supersedes pending editor requests.""" + if self.predecessor_saves and self.client_save_id is None: + raise ValueError("predecessor_saves requires client_save_id") + predecessor_ids = [predecessor.client_save_id for predecessor in self.predecessor_saves] + if len(set(predecessor_ids)) != len(predecessor_ids) or self.client_save_id in predecessor_ids: + raise ValueError("predecessor save ids must be unique and differ from client_save_id") + return self + class BaseWorkflowRunPayload(BaseModel): files: list[dict[str, Any]] | None = Field(default=None) @@ -325,6 +356,9 @@ class WorkflowResponse(ResponseModel): validation_alias=AliasChoices("features_dict", "features"), ) hash: str = Field(validation_alias=AliasChoices("unique_hash", "hash")) + client_save_id: UUID | None = Field( + default=None, validation_alias=AliasChoices("draft_client_save_id", "client_save_id") + ) version: str # NULL for drafts and for versions published before numbering was introduced; those # render as "Untitled Version" instead of `#N`. Never 0, so clients must test for null. @@ -416,6 +450,7 @@ class WorkflowPublishResponse(ResponseModel): class SyncDraftWorkflowResponse(ResponseModel): result: str hash: str + client_save_id: UUID | None = None updated_at: int @@ -692,6 +727,25 @@ class DraftWorkflowApi(Resource): variable_factory.build_conversation_variable_from_mapping(obj) for obj in args_model.conversation_variables ] + predecessor_saves = [] + for predecessor in args_model.predecessor_saves: + predecessor_patch = predecessor.environment_variable_patch + predecessor_saves.append( + DraftSavePredecessor( + client_save_id=str(predecessor.client_save_id), + environment_variable_upserts=[ + variable_factory.build_environment_variable_from_mapping(obj) + for obj in Workflow.normalize_environment_variable_mappings( + predecessor_patch.environment_variables + ) + ] + if predecessor_patch is not None + else (), + deleted_environment_variable_ids=predecessor_patch.deleted_environment_variable_ids + if predecessor_patch is not None + else (), + ) + ) # Keep this accepted snapshot while the post-commit handoff may wait. with session_factory.create_session() as write_session: workflow = workflow_service.sync_draft_workflow( @@ -699,6 +753,11 @@ class DraftWorkflowApi(Resource): graph=args_model.graph, features=args_model.features, unique_hash=args_model.hash, + client_save_id=str(args_model.client_save_id) if args_model.client_save_id is not None else None, + base_client_save_id=str(args_model.base_client_save_id) + if args_model.base_client_save_id is not None + else None, + predecessor_saves=predecessor_saves, account=current_user, environment_variables=[], conversation_variables=conversation_variables, @@ -725,6 +784,7 @@ class DraftWorkflowApi(Resource): { "result": "success", "hash": workflow.unique_hash, + "client_save_id": workflow.draft_client_save_id, "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at), }, ) diff --git a/api/migrations/versions/2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py b/api/migrations/versions/2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py new file mode 100644 index 00000000000..447a6862cdc --- /dev/null +++ b/api/migrations/versions/2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py @@ -0,0 +1,30 @@ +"""Track the last editor draft save for unload-save conflict protection. + +Revision ID: 8e06b3f72457 +Revises: 7d95a2e61346 +Create Date: 2026-11-03 12:00:00.000000 +""" + +import sqlalchemy as sa +from alembic import op + +from models.types import StringUUID + +revision = "8e06b3f72457" +down_revision = "7d95a2e61346" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("workflows") as batch_op: + batch_op.add_column(sa.Column("draft_client_save_id", StringUUID(), nullable=True)) + batch_op.add_column(sa.Column("draft_client_save_account_id", StringUUID(), nullable=True)) + batch_op.add_column(sa.Column("draft_client_save_hash", sa.String(64), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("workflows") as batch_op: + batch_op.drop_column("draft_client_save_hash") + batch_op.drop_column("draft_client_save_account_id") + batch_op.drop_column("draft_client_save_id") diff --git a/api/models/workflow.py b/api/models/workflow.py index 9ee6420131d..5b2b1b7c1f1 100644 --- a/api/models/workflow.py +++ b/api/models/workflow.py @@ -230,6 +230,9 @@ class Workflow(TypeBase): marked_name: Mapped[str] = mapped_column(String(255), default="") marked_comment: Mapped[str] = mapped_column(String(255), default="") graph: Mapped[str] = mapped_column(LongText, default=None) + draft_client_save_id: Mapped[str | None] = mapped_column(StringUUID, default=None) + draft_client_save_account_id: Mapped[str | None] = mapped_column(StringUUID, default=None) + draft_client_save_hash: Mapped[str | None] = mapped_column(String(64), default=None) _features: Mapped[Any] = mapped_column("features", LongText, default=None) created_by: Mapped[str] = mapped_column(StringUUID, nullable=False, default=None) created_at: Mapped[datetime] = mapped_column( diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md index bf06cf817a5..8418dfd40fa 100644 --- a/api/openapi/markdown/console-openapi.md +++ b/api/openapi/markdown/console-openapi.md @@ -35191,6 +35191,13 @@ Verification of the plugin. | ---- | ---- | ----------- | -------- | | PreProcessingRuleKey | string | | | +#### PredecessorDraftSavePayload + +| Name | Type | Description | Required | +| ---- | ---- | ----------- | -------- | +| client_save_id | string (uuid) | | Yes | +| environment_variable_patch | [SyncEnvironmentVariablePatchPayload](#syncenvironmentvariablepatchpayload) | | No | + #### PreflightContextCard | Name | Type | Description | Required | @@ -37375,6 +37382,7 @@ Query parameters for listing snippet published workflows. | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | +| client_save_id | string | | No | | conversation_variables | [ [WorkflowConversationVariableResponse](#workflowconversationvariableresponse) ] | | Yes | | created_at | integer | | Yes | | created_by | [SimpleAccountResponse](#simpleaccountresponse) | | No | @@ -37579,17 +37587,21 @@ The subscription constructor of the trigger provider | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | | _is_collaborative | boolean | | No | +| base_client_save_id | string | Receipt ID returned with the editor's base draft. | No | +| client_save_id | string | Unique ID for this editor save attempt. | No | | conversation_variables | [ object ] | | No | | environment_variable_patch | [SyncEnvironmentVariablePatchPayload](#syncenvironmentvariablepatchpayload) | | No | | features | object | | Yes | | force | boolean | Explicitly confirm saving an empty workflow graph. | No | | graph | object | | Yes | | hash | string | | No | +| predecessor_saves | [ [PredecessorDraftSavePayload](#predecessordraftsavepayload) ] | Pending saves in editor order, including inherited patches; requires client_save_id. | No | #### SyncDraftWorkflowResponse | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | +| client_save_id | string | | No | | hash | string | | Yes | | result | string | | Yes | | updated_at | integer | | Yes | @@ -39975,6 +39987,7 @@ Data entity | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | +| client_save_id | string | | No | | conversation_variables | [ [WorkflowConversationVariableResponse](#workflowconversationvariableresponse) ] | | Yes | | created_at | integer | | Yes | | created_by | [SimpleAccountResponse](#simpleaccountresponse) | | No | diff --git a/api/repositories/workflow_draft_sync_repository.py b/api/repositories/workflow_draft_sync_repository.py index db4fb259f9c..2d98963a481 100644 --- a/api/repositories/workflow_draft_sync_repository.py +++ b/api/repositories/workflow_draft_sync_repository.py @@ -56,6 +56,7 @@ return 0 class ServerDraftChange(BaseModel): revision: str hash: str + client_save_id: str | None = None updated_at: int graph: dict[str, object] previous_graph: dict[str, object] | None = None diff --git a/api/services/workflow_collaboration_service.py b/api/services/workflow_collaboration_service.py index 8e1f0c8a01a..4447cafdf07 100644 --- a/api/services/workflow_collaboration_service.py +++ b/api/services/workflow_collaboration_service.py @@ -13,9 +13,11 @@ from sqlalchemy import select from sqlalchemy.orm import Session from configs import dify_config +from core.db.session_factory import session_factory from core.rbac import RBACPermission, RBACResourceScope from models.account import Account from models.model import App +from models.workflow import Workflow from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository, WorkflowSessionInfo from services.enterprise.rbac_service import RBACService @@ -147,7 +149,12 @@ class WorkflowCollaborationService: self._socketio.enter_room(sid, workflow_id) self.broadcast_online_users(workflow_id) - self._socketio.emit("status", {"isLeader": is_leader}, room=sid) + status: dict[str, object] = {"isLeader": is_leader} + if is_leader: + baseline = self._get_draft_baseline(workflow_id, session=session) + if baseline is not None: + status["draftBaseline"] = baseline + self._socketio.emit("status", status, room=sid) return str(user_id), is_leader @@ -185,7 +192,12 @@ class WorkflowCollaborationService: self.broadcast_online_users(workflow_id) def relay_collaboration_event(self, sid: str, data: Mapping[str, object]) -> tuple[dict[str, object], int]: - """Route collaboration control events, directing save and graph-resync requests to the active leader.""" + """Route authenticated collaboration events to the active leader or room. + + Successful save acknowledgments retain the persisted hash and save receipt + together so a requester can safely become the next saver. Invalid or failed + acknowledgments are rejected without reporting a successful save. + """ mapping = self._repository.get_sid_mapping(sid) if not mapping: return {"msg": "unauthorized"}, 401 @@ -290,10 +302,13 @@ class WorkflowCollaborationService: workflow_hash = result.get("hash") updated_at = result.get("updatedAt") + client_save_id = result.get("clientSaveId") if not isinstance(workflow_hash, str) or not workflow_hash: return {"msg": "invalid_sync_response", "requestId": request_id}, 502 if not isinstance(updated_at, int) or isinstance(updated_at, bool): return {"msg": "invalid_sync_response", "requestId": request_id}, 502 + if client_save_id is not None and not isinstance(client_save_id, str): + return {"msg": "invalid_sync_response", "requestId": request_id}, 502 return { "msg": "workflow_synced", @@ -301,6 +316,7 @@ class WorkflowCollaborationService: "success": True, "hash": workflow_hash, "updatedAt": updated_at, + **({"clientSaveId": client_save_id} if "clientSaveId" in result else {}), }, 200 if event_type == "graph_resync_request": @@ -388,10 +404,34 @@ class WorkflowCollaborationService: for sid in self._repository.get_session_sids(workflow_id): try: is_leader = new_leader_sid is not None and sid == new_leader_sid - self._socketio.emit("status", {"isLeader": is_leader}, room=sid) + status: dict[str, object] = {"isLeader": is_leader} + if is_leader: + with session_factory.create_session() as session: + baseline = self._get_draft_baseline(workflow_id, session=session) + if baseline is not None: + status["draftBaseline"] = baseline + self._socketio.emit("status", status, room=sid) except Exception: logging.exception("Failed to emit leader status to session %s", sid) + @staticmethod + def _get_draft_baseline(workflow_id: str, *, session: Session) -> dict[str, str | None] | None: + """Read matching hash and receipt for an already-authorized App room. + + Join the complete App/tenant owner because the promoted socket can belong + to another worker. End the read transaction before emitting its baseline; + the next save still validates this receipt under the draft row lock. + """ + with session.begin(): + workflow = session.scalar( + select(Workflow) + .join(App, (Workflow.app_id == App.id) & (Workflow.tenant_id == App.tenant_id)) + .where(App.id == workflow_id, App.status == "normal", Workflow.version == Workflow.VERSION_DRAFT) + ) + if workflow is None: + return None + return {"hash": workflow.unique_hash, "client_save_id": workflow.draft_client_save_id} + def get_current_leader(self, workflow_id: str) -> str | None: return self._repository.get_current_leader(workflow_id) diff --git a/api/services/workflow_draft_sync_service.py b/api/services/workflow_draft_sync_service.py index 5103557a762..526983e1291 100644 --- a/api/services/workflow_draft_sync_service.py +++ b/api/services/workflow_draft_sync_service.py @@ -33,6 +33,7 @@ def _change_from_workflow(workflow: Workflow, previous_graph: dict[str, object] return ServerDraftChange( revision=f"{updated_at.isoformat(timespec='microseconds')}:{workflow.unique_hash}", hash=workflow.unique_hash, + client_save_id=workflow.draft_client_save_id, updated_at=int(updated_at.timestamp()), graph=dict(workflow.graph_dict), previous_graph=previous_graph, @@ -150,6 +151,7 @@ class WorkflowDraftSyncService: return { "revision": change.revision, "hash": change.hash, + "client_save_id": change.client_save_id, "updated_at": change.updated_at, "update": base64.b64decode(change.update), } diff --git a/api/services/workflow_service.py b/api/services/workflow_service.py index 42a3ebafa04..f52f20f2da4 100644 --- a/api/services/workflow_service.py +++ b/api/services/workflow_service.py @@ -182,6 +182,15 @@ HumanInputNode = _DebugHumanInputNode _file_access_controller = DatabaseFileAccessController() +@dataclass(frozen=True) +class DraftSavePredecessor: + """A pending editor save and its environment patch, ordered before the new save.""" + + client_save_id: str + environment_variable_upserts: Sequence[VariableBase] = () + deleted_environment_variable_ids: Sequence[str] = () + + def _merge_environment_variable_patch( current_variables: Sequence[VariableBase], environment_variable_upserts: Sequence[VariableBase], @@ -472,6 +481,9 @@ class WorkflowService: sync_agent_bindings: bool = True, graph_only: bool = False, require_empty_graph_confirmation: bool = False, + client_save_id: str | None = None, + base_client_save_id: str | None = None, + predecessor_saves: Sequence[DraftSavePredecessor] = (), ) -> Workflow: """ Sync draft workflow. @@ -495,17 +507,69 @@ class WorkflowService: Builder repair/rollback clears; public force=true uses the same overload. Neither overload bypasses the existing draft hash check. + An unload save may supersede its own pending predecessor without waiting + for the predecessor response. The locked draft must still be the exact + result of a named save by this account. A changed graph or another save + invalidates that receipt. Untagged saves receive a server ID; receipt and graph + commit together, after Agent bindings have materialized the final graph. + Inherited environment patches through the matched predecessor are already + committed and must not be replayed over dedicated variable edits. Without + a matching predecessor, tagged saves require the unchanged base receipt, + including when undo or variable-only edits restore the same graph hash. + :raises WorkflowHashNotEqualError :raises EmptyWorkflowConfirmationRequiredError """ if environment_variable_upserts is None and deleted_environment_variable_ids: raise ValueError("Deleted environment variable ids require an environment variable patch.") + if predecessor_saves and client_save_id is None: + raise ValueError("predecessor_saves requires client_save_id") # fetch draft workflow by app_model workflow = self.get_draft_workflow_for_creation(app_model=app_model, session=session) - if workflow and workflow.unique_hash != unique_hash: - raise WorkflowHashNotEqualError() + committed_predecessor_index = -1 + if workflow: + current_hash = workflow.unique_hash + if workflow.draft_client_save_account_id == account.id and workflow.draft_client_save_hash == current_hash: + committed_predecessor_index = next( + ( + index + for index, predecessor in enumerate(predecessor_saves) + if predecessor.client_save_id == workflow.draft_client_save_id + ), + -1, + ) + if current_hash != unique_hash and committed_predecessor_index < 0: + raise WorkflowHashNotEqualError() + if ( + committed_predecessor_index < 0 + and client_save_id is not None + and workflow.draft_client_save_id != base_client_save_id + ): + raise WorkflowHashNotEqualError() + + pending_predecessors = predecessor_saves[committed_predecessor_index + 1 :] + if any( + predecessor.environment_variable_upserts or predecessor.deleted_environment_variable_ids + for predecessor in pending_predecessors + ): + combined_upserts: dict[str, VariableBase] = {} + combined_deletions: set[str] = set() + patches = [ + (predecessor.environment_variable_upserts, predecessor.deleted_environment_variable_ids) + for predecessor in pending_predecessors + ] + patches.append((environment_variable_upserts or (), deleted_environment_variable_ids)) + for upserts, deleted_ids in patches: + for variable in upserts: + combined_upserts[variable.id] = variable + combined_deletions.discard(variable.id) + for variable_id in deleted_ids: + combined_upserts.pop(variable_id, None) + combined_deletions.add(variable_id) + environment_variable_upserts = list(combined_upserts.values()) + deleted_environment_variable_ids = sorted(combined_deletions) if require_empty_graph_confirmation and graph.get("nodes") == [] and graph.get("edges") == []: previous_graph = workflow.graph_dict if workflow is not None else None @@ -578,6 +642,11 @@ class WorkflowService: draft_workflow=workflow, ) + workflow.draft_client_save_id = client_save_id or str(uuid.uuid4()) + workflow.draft_client_save_account_id = account.id + workflow.draft_client_save_hash = workflow.unique_hash + workflow.updated_at = naive_utc_now() + coordination_id = coordination_id or str(uuid.uuid4()) WorkflowDraftCoordinationService.stage( session, diff --git a/api/tests/e2e_tests/containers/repositories/test_workflow_draft_sync_repository.py b/api/tests/e2e_tests/containers/repositories/test_workflow_draft_sync_repository.py index 8112ed14460..d7d34fa722b 100644 --- a/api/tests/e2e_tests/containers/repositories/test_workflow_draft_sync_repository.py +++ b/api/tests/e2e_tests/containers/repositories/test_workflow_draft_sync_repository.py @@ -23,7 +23,9 @@ from repositories.workflow_draft_sync_repository import ServerDraftChange, Workf def test_only_the_current_lease_can_accept_one_update_for_the_current_revision() -> None: repository = WorkflowDraftSyncRepository() app_id = uuid.uuid4().hex - first = ServerDraftChange(revision="1", hash="h1", updated_at=1, graph={"nodes": [], "edges": []}) + first = ServerDraftChange( + revision="1", hash="h1", client_save_id=str(uuid.uuid4()), updated_at=1, graph={"nodes": [], "edges": []} + ) assert repository.announce(app_id, first) assert repository.claim(app_id, "old") assert not repository.claim(app_id, "new") @@ -34,7 +36,9 @@ def test_only_the_current_lease_can_accept_one_update_for_the_current_revision() assert not repository.claim(app_id, "third") assert not repository.accept(app_id, "old", "1", "b2xk") - second = first.model_copy(update={"revision": "2", "hash": "h2", "updated_at": 2}) + second = first.model_copy( + update={"revision": "2", "hash": "h2", "updated_at": 2, "client_save_id": str(uuid.uuid4())} + ) assert repository.announce(app_id, second) assert not repository.accept(app_id, "new", "1", "b2xk") assert repository.accept(app_id, "new", "2", "bmV3") @@ -43,6 +47,7 @@ def test_only_the_current_lease_can_accept_one_update_for_the_current_revision() stored = repository.get(app_id) assert stored is not None assert stored.revision == "2" + assert stored.client_save_id == second.client_save_id assert stored.update == "bmV3" assert stored.graph == {"nodes": [], "edges": []} @@ -116,7 +121,9 @@ def test_cloud_draft_sync_accepts_and_clears_updates_on_cluster_with_builder_dis monkeypatch.setattr(repo_module, "redis_client", redis) repository = WorkflowDraftSyncRepository() app_id = str(uuid.uuid4()) - change = ServerDraftChange(revision="1", hash="h1", updated_at=1, graph={"nodes": [], "edges": []}) + change = ServerDraftChange( + revision="1", hash="h1", client_save_id=str(uuid.uuid4()), updated_at=1, graph={"nodes": [], "edges": []} + ) try: assert repository.announce(app_id, change) diff --git a/api/tests/e2e_tests/containers/services/test_workflow_draft_client_save.py b/api/tests/e2e_tests/containers/services/test_workflow_draft_client_save.py new file mode 100644 index 00000000000..1aeba254683 --- /dev/null +++ b/api/tests/e2e_tests/containers/services/test_workflow_draft_client_save.py @@ -0,0 +1,489 @@ +"""Unload saves supersede only the same editor's accepted pending save. + +The real POST parser, service and PostgreSQL transactions exercise both request +orders. Authentication decorators are unwrapped to supply persisted authorized +owners, matching the existing draft consistency tests. +""" + +import inspect +import json +from uuid import uuid4 + +import pytest +from flask import Flask +from sqlalchemy.orm import Session, sessionmaker + +from controllers.console.app import workflow as workflow_api +from controllers.console.app.error import DraftWorkflowNotSync +from graphon.variables import StringVariable +from models.account import Account, TenantAccountJoin, TenantAccountRole +from models.workflow import Workflow +from services.workflow_service import WorkflowService +from tests.workflow_empty_draft_test_utils import POPULATED_GRAPH, DraftOwner, seed_owner + + +def _graph(title: str) -> dict[str, object]: + return {"nodes": [{"id": "start", "data": {"type": "start", "title": title}}], "edges": []} + + +def _post( + app: Flask, + owner: DraftOwner, + *, + title: str, + unique_hash: str, + save_id: str | None, + predecessors: tuple[str, ...] = (), + base_save_id: str | None = None, + predecessor_patches: dict[str, dict[str, object]] | None = None, + environment_patch: dict[str, object] | None = None, + collaborative: bool = False, + content_type: str = "application/json", + account: Account | None = None, +) -> dict[str, object]: + payload = { + "graph": _graph(title), + "features": {}, + "hash": unique_hash, + "client_save_id": save_id, + "base_client_save_id": base_save_id, + "predecessor_saves": [ + { + "client_save_id": predecessor_id, + "environment_variable_patch": (predecessor_patches or {}).get(predecessor_id), + } + for predecessor_id in predecessors + ], + "environment_variable_patch": environment_patch, + "_is_collaborative": collaborative, + } + api = workflow_api.DraftWorkflowApi() + with app.test_request_context( + f"/apps/{owner.app.id}/workflows/draft", + method="POST", + data=json.dumps(payload), + content_type=content_type, + ): + return inspect.unwrap(api.post)(api, account or owner.account, app_model=owner.app) + + +def _read(session: Session, owner: DraftOwner) -> Workflow: + assert owner.draft is not None + with Session(session.get_bind(), expire_on_commit=False) as reader: + draft = reader.get(Workflow, owner.draft.id) + assert draft is not None + return draft + + +@pytest.mark.parametrize("collaborative", [False, True], ids=["regular", "collaborative"]) +@pytest.mark.parametrize("content_type", ["application/json", "text/plain"], ids=["json", "beacon"]) +@pytest.mark.parametrize("unload_first", [False, True], ids=["autosave-first", "unload-first"]) +def test_unload_keeps_latest_graph_in_both_arrival_orders( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + collaborative: bool, + content_type: str, + unload_first: bool, +) -> None: + session = db_session_with_containers + assert session.get_bind().dialect.name == "postgresql" + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + autosave_id, unload_id = str(uuid4()), str(uuid4()) + + def autosave() -> None: + _post( + flask_app_with_containers, + owner, + title="Autosave A", + unique_hash=initial_hash, + save_id=autosave_id, + collaborative=collaborative, + ) + + def unload() -> dict[str, object]: + return _post( + flask_app_with_containers, + owner, + title="Latest edit B", + unique_hash=initial_hash, + save_id=unload_id, + predecessors=(str(uuid4()), autosave_id), + collaborative=collaborative, + content_type=content_type, + ) + + if unload_first: + response = unload() + with pytest.raises(DraftWorkflowNotSync): + autosave() + else: + autosave() + response = unload() + + saved = _read(session, owner) + assert saved.graph_dict == _graph("Latest edit B") + assert saved.unique_hash == response["hash"] + assert saved.draft_client_save_id == unload_id + assert response["client_save_id"] == unload_id + assert saved.draft_client_save_account_id == owner.account.id + assert saved.draft_client_save_hash == saved.unique_hash + assert [(value.name, value.value) for value in saved.environment_variables] == [("SERVER_VALUE", "keep-env")] + assert [(value.name, value.value) for value in saved.conversation_variables] == [("topic", "keep-conversation")] + assert json.loads(saved.serialized_features) == {"file_upload": {"enabled": False}} + + +@pytest.mark.parametrize("collaborative", [False, True], ids=["regular", "collaborative"]) +@pytest.mark.parametrize("writer", ["tagged", "untagged", "direct"], ids=["another-editor", "untagged-save", "builder"]) +def test_intervening_writer_prevents_stale_unload_overwrite( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + collaborative: bool, + writer: str, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + autosave_id = str(uuid4()) + response = _post( + flask_app_with_containers, owner, title="Autosave A", unique_hash=initial_hash, save_id=autosave_id + ) + assert isinstance(response["hash"], str) + if writer == "direct": + with Session(session.get_bind()) as external, external.begin(): + draft = external.get(Workflow, owner.draft.id) + assert draft is not None + draft.graph = json.dumps(_graph("Builder edit")) + else: + _post( + flask_app_with_containers, + owner, + title="Autosave A", + unique_hash=response["hash"], + save_id=str(uuid4()) if writer == "tagged" else None, + base_save_id=autosave_id, + ) + before = _read(session, owner) + if writer == "untagged": + assert before.draft_client_save_id is not None + assert before.draft_client_save_id != autosave_id + assert before.draft_client_save_account_id == owner.account.id + assert before.draft_client_save_hash == before.unique_hash + with pytest.raises(DraftWorkflowNotSync): + _post( + flask_app_with_containers, + owner, + title="Stale unload", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id,), + collaborative=collaborative, + ) + after = _read(session, owner) + assert after.graph == before.graph + assert after.draft_client_save_id == before.draft_client_save_id + + +def test_predecessor_receipt_is_bound_to_the_saving_account( + flask_app_with_containers: Flask, + db_session_with_containers: Session, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + autosave_id = str(uuid4()) + _post(flask_app_with_containers, owner, title="Autosave A", unique_hash=initial_hash, save_id=autosave_id) + other_account = Account(name="Collaborator", email=f"{uuid4()}@example.invalid") + session.add(other_account) + session.flush() + session.add( + TenantAccountJoin( + tenant_id=owner.app.tenant_id, + account_id=other_account.id, + role=TenantAccountRole.EDITOR, + current=True, + ) + ) + session.commit() + with pytest.raises(DraftWorkflowNotSync): + _post( + flask_app_with_containers, + owner, + title="Other account stale edit", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id,), + collaborative=True, + account=other_account, + ) + assert _read(session, owner).graph_dict == _graph("Autosave A") + + +def test_receipt_rolls_back_with_the_graph(db_session_with_containers: Session) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + service = WorkflowService(sessionmaker(bind=session.get_bind(), expire_on_commit=False)) + with Session(session.get_bind()) as writer: + staged = service.sync_draft_workflow( + app_model=owner.app, + graph=_graph("Rolled back"), + features={}, + unique_hash=initial_hash, + account=owner.account, + environment_variables=[], + conversation_variables=[], + session=writer, + graph_only=True, + commit=False, + client_save_id=str(uuid4()), + ) + writer.flush() + assert staged.draft_client_save_hash == staged.unique_hash + writer.rollback() + saved = _read(session, owner) + assert saved.unique_hash == initial_hash + assert saved.draft_client_save_id is None + assert saved.draft_client_save_account_id is None + assert saved.draft_client_save_hash is None + + +@pytest.mark.parametrize("changes_graph", [False, True], ids=["variable-only-autosave", "graph-autosave"]) +def test_inherited_committed_patch_does_not_overwrite_later_dedicated_variable_edit( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + changes_graph: bool, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + original = owner.draft.environment_variables[0] + patch_a = {"environment_variables": [original.model_copy(update={"value": "A"}).model_dump(mode="json")]} + autosave_id, queued_id = str(uuid4()), str(uuid4()) + _post( + flask_app_with_containers, + owner, + title="Autosave A" if changes_graph else "Start", + unique_hash=initial_hash, + save_id=autosave_id, + environment_patch=patch_a, + ) + with Session(session.get_bind()) as writer: + WorkflowService().patch_draft_workflow_environment_variables( + app_model=owner.app, + environment_variables=[original.model_copy(update={"value": "Dedicated C"})], + deleted_environment_variable_ids=[], + account=owner.account, + session=writer, + ) + queued_variable = StringVariable(id=str(uuid4()), name="QUEUED", value="Q") + current_variable = StringVariable(id=str(uuid4()), name="CURRENT", value="B") + response = _post( + flask_app_with_containers, + owner, + title="Unload B", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id, queued_id), + predecessor_patches={ + autosave_id: patch_a, + queued_id: {"environment_variables": [queued_variable.model_dump(mode="json")]}, + }, + environment_patch={"environment_variables": [current_variable.model_dump(mode="json")]}, + collaborative=True, + ) + saved = _read(session, owner) + assert saved.graph_dict == _graph("Unload B") + assert {variable.name: variable.value for variable in saved.environment_variables} == { + "SERVER_VALUE": "Dedicated C", + "QUEUED": "Q", + "CURRENT": "B", + } + api = workflow_api.DraftWorkflowApi() + with Session(session.get_bind()) as reader: + draft_response = inspect.unwrap(api.get)(api, reader, app_model=owner.app) + assert draft_response["client_save_id"] == response["client_save_id"] + + +@pytest.mark.parametrize("has_base_receipt", [False, True], ids=["initial-base", "acknowledged-base"]) +def test_uncommitted_inherited_patches_apply_in_order_from_unchanged_base( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + has_base_receipt: bool, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + base_id = None + if has_base_receipt: + base = _post(flask_app_with_containers, owner, title="Start", unique_hash=initial_hash, save_id=None) + assert isinstance(base["client_save_id"], str) + base_id = base["client_save_id"] + original = owner.draft.environment_variables[0] + queued_variable = StringVariable(id=str(uuid4()), name="QUEUED", value="Q") + autosave_id, queued_id = str(uuid4()), str(uuid4()) + _post( + flask_app_with_containers, + owner, + title="Unload B", + unique_hash=initial_hash, + save_id=str(uuid4()), + base_save_id=base_id, + predecessors=(autosave_id, queued_id), + predecessor_patches={ + autosave_id: {"deleted_environment_variable_ids": [original.id]}, + queued_id: { + "environment_variables": [ + original.model_copy(update={"value": "Queued replacement"}).model_dump(mode="json"), + queued_variable.model_dump(mode="json"), + ] + }, + }, + environment_patch={"deleted_environment_variable_ids": [queued_variable.id]}, + ) + saved = _read(session, owner) + assert saved.graph_dict == _graph("Unload B") + assert [(variable.name, variable.value) for variable in saved.environment_variables] == [ + ("SERVER_VALUE", "Queued replacement") + ] + + +@pytest.mark.parametrize("tagged_writer", [False, True], ids=["untagged", "tagged"]) +def test_same_graph_intervening_save_rejects_ambiguous_inherited_patch( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + tagged_writer: bool, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + original = owner.draft.environment_variables[0] + autosave_id = str(uuid4()) + patch_a = {"environment_variables": [original.model_copy(update={"value": "A"}).model_dump(mode="json")]} + autosave = _post( + flask_app_with_containers, + owner, + title="Start", + unique_hash=initial_hash, + save_id=autosave_id, + environment_patch=patch_a, + ) + assert isinstance(autosave["client_save_id"], str) + _post( + flask_app_with_containers, + owner, + title="Start", + unique_hash=initial_hash, + save_id=str(uuid4()) if tagged_writer else None, + base_save_id=autosave["client_save_id"], + environment_patch={ + "environment_variables": [original.model_copy(update={"value": "Newer C"}).model_dump(mode="json")] + }, + ) + with pytest.raises(DraftWorkflowNotSync): + _post( + flask_app_with_containers, + owner, + title="Unload B", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id,), + predecessor_patches={autosave_id: patch_a}, + collaborative=True, + ) + saved = _read(session, owner) + assert saved.unique_hash == initial_hash + assert saved.environment_variables[0].value == "Newer C" + + +@pytest.mark.parametrize("unload_first", [False, True], ids=["autosave-first", "unload-first"]) +def test_variable_only_saves_keep_latest_patch_in_both_arrival_orders( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + unload_first: bool, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + original = owner.draft.environment_variables[0] + autosave_id = str(uuid4()) + patch_a = {"environment_variables": [original.model_copy(update={"value": "A"}).model_dump(mode="json")]} + + def autosave() -> None: + _post( + flask_app_with_containers, + owner, + title="Start", + unique_hash=initial_hash, + save_id=autosave_id, + environment_patch=patch_a, + ) + + def unload() -> None: + _post( + flask_app_with_containers, + owner, + title="Start", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id,), + predecessor_patches={autosave_id: patch_a}, + environment_patch={ + "environment_variables": [original.model_copy(update={"value": "B"}).model_dump(mode="json")] + }, + collaborative=True, + ) + + if unload_first: + unload() + with pytest.raises(DraftWorkflowNotSync): + autosave() + else: + autosave() + unload() + saved = _read(session, owner) + assert saved.unique_hash == initial_hash + assert saved.environment_variables[0].value == "B" + + +@pytest.mark.parametrize("collaborative", [False, True], ids=["regular", "collaborative"]) +def test_undo_unload_with_original_graph_hash_rejects_late_autosave( + flask_app_with_containers: Flask, + db_session_with_containers: Session, + collaborative: bool, +) -> None: + session = db_session_with_containers + owner = seed_owner(session, graph=POPULATED_GRAPH) + assert owner.draft is not None + initial_hash = owner.draft.unique_hash + autosave_id = str(uuid4()) + response = _post( + flask_app_with_containers, + owner, + title="Start", + unique_hash=initial_hash, + save_id=str(uuid4()), + predecessors=(autosave_id,), + collaborative=collaborative, + ) + assert response["hash"] == initial_hash + with pytest.raises(DraftWorkflowNotSync): + _post( + flask_app_with_containers, + owner, + title="Earlier edit before undo", + unique_hash=initial_hash, + save_id=autosave_id, + collaborative=collaborative, + ) + saved = _read(session, owner) + assert saved.graph_dict == POPULATED_GRAPH + assert saved.draft_client_save_id == response["client_save_id"] diff --git a/api/tests/unit_tests/controllers/console/app/test_workflow.py b/api/tests/unit_tests/controllers/console/app/test_workflow.py index 119eb3d6845..76a420f2141 100644 --- a/api/tests/unit_tests/controllers/console/app/test_workflow.py +++ b/api/tests/unit_tests/controllers/console/app/test_workflow.py @@ -10,6 +10,7 @@ from datetime import datetime from inspect import unwrap from types import SimpleNamespace from unittest.mock import ANY, MagicMock, Mock +from uuid import UUID, uuid4 import pytest from flask import Flask @@ -422,7 +423,9 @@ class TestWorkflowCases: graph: dict[str, list[dict[str, str]]], confirmation: dict[str, bool], ) -> None: - workflow = SimpleNamespace(unique_hash="next-hash", updated_at=None, created_at=datetime(2024, 1, 1)) + workflow = SimpleNamespace( + unique_hash="next-hash", draft_client_save_id=None, updated_at=None, created_at=datetime(2024, 1, 1) + ) sync_draft_workflow = Mock(return_value=workflow) monkeypatch.setattr( workflow_module, @@ -582,6 +585,52 @@ class TestWorkflowCases: } ) + @staticmethod + @pytest.mark.parametrize( + "save_fields", + [ + {"client_save_id": "not-a-uuid"}, + {"base_client_save_id": "not-a-uuid"}, + {"client_save_id": str(uuid4()), "predecessor_saves": [{"client_save_id": "not-a-uuid"}]}, + {"predecessor_saves": [{"client_save_id": str(uuid4())}]}, + ], + ids=["invalid-save-id", "invalid-base-id", "invalid-predecessor-id", "missing-save-id"], + ) + def test_sync_draft_workflow_rejects_invalid_save_receipt(save_fields: dict[str, object]) -> None: + with pytest.raises(ValidationError): + workflow_module.SyncDraftWorkflowPayload.model_validate({"graph": {}, "features": {}, **save_fields}) + + @staticmethod + def test_sync_draft_workflow_accepts_pending_predecessor_ids() -> None: + save_id = uuid4() + predecessor_ids = [uuid4(), uuid4()] + payload = workflow_module.SyncDraftWorkflowPayload.model_validate( + { + "graph": {}, + "features": {}, + "client_save_id": str(save_id).upper(), + "predecessor_saves": [{"client_save_id": str(save_id)} for save_id in predecessor_ids], + } + ) + assert payload.client_save_id == save_id + assert isinstance(payload.client_save_id, UUID) + assert [predecessor.client_save_id for predecessor in payload.predecessor_saves] == predecessor_ids + + @staticmethod + @pytest.mark.parametrize("duplicate", [False, True], ids=["self-reference", "duplicate"]) + def test_sync_draft_workflow_rejects_ambiguous_predecessor_order(duplicate: bool) -> None: + save_id, predecessor_id = str(uuid4()), str(uuid4()) + predecessor_ids = [predecessor_id, predecessor_id] if duplicate else [save_id] + with pytest.raises(ValidationError, match="must be unique and differ"): + workflow_module.SyncDraftWorkflowPayload.model_validate( + { + "graph": {}, + "features": {}, + "client_save_id": save_id, + "predecessor_saves": [{"client_save_id": predecessor_id} for predecessor_id in predecessor_ids], + } + ) + @staticmethod def test_sync_draft_workflow_hash_mismatch(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: def _raise(*_args, **_kwargs): diff --git a/api/tests/unit_tests/migrations/versions/test_2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py b/api/tests/unit_tests/migrations/versions/test_2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py new file mode 100644 index 00000000000..d6d5c1752dd --- /dev/null +++ b/api/tests/unit_tests/migrations/versions/test_2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py @@ -0,0 +1,63 @@ +"""The unload-save receipt migrates without changing existing workflow graphs.""" + +import importlib.util +from io import StringIO +from pathlib import Path +from types import ModuleType + +import pytest +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import Column, MetaData, String, Table, create_engine, inspect, select + + +def _migration() -> ModuleType: + path = ( + Path(__file__).parents[4] / "migrations/versions/2026_11_03_1200-8e06b3f72457_add_workflow_draft_client_save.py" + ) + spec = importlib.util.spec_from_file_location("workflow_draft_client_save_migration", path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.mark.parametrize("dialect", ["postgresql", "mysql"]) +def test_receipt_migration_renders_for_supported_databases(dialect: str) -> None: + output = StringIO() + with Operations.context( + MigrationContext.configure(dialect_name=dialect, opts={"as_sql": True, "output_buffer": output}) + ): + _migration().upgrade() + _migration().downgrade() + sql = output.getvalue() + for column in ("draft_client_save_id", "draft_client_save_account_id", "draft_client_save_hash"): + assert f"ADD COLUMN {column}" in sql + assert f"DROP COLUMN {column}" in sql + + +def test_receipt_upgrade_and_downgrade_preserve_existing_workflow() -> None: + engine = create_engine("sqlite://") + metadata = MetaData() + workflows = Table("workflows", metadata, Column("id", String, primary_key=True), Column("graph", String)) + metadata.create_all(engine) + with engine.begin() as connection: + connection.execute(workflows.insert().values(id="existing", graph='{"nodes": [], "edges": []}')) + with Operations.context(MigrationContext.configure(connection)): + _migration().upgrade() + upgraded = Table("workflows", MetaData(), autoload_with=connection) + assert connection.execute(select(upgraded)).one()._asdict() == { + "id": "existing", + "graph": '{"nodes": [], "edges": []}', + "draft_client_save_id": None, + "draft_client_save_account_id": None, + "draft_client_save_hash": None, + } + _migration().downgrade() + assert {column["name"] for column in inspect(connection).get_columns("workflows")} == {"id", "graph"} + assert connection.execute(select(workflows)).one()._asdict() == { + "id": "existing", + "graph": '{"nodes": [], "edges": []}', + } + engine.dispose() diff --git a/api/tests/unit_tests/repositories/test_workflow_draft_sync_repository.py b/api/tests/unit_tests/repositories/test_workflow_draft_sync_repository.py index b0fae8a6a2d..a738045dd10 100644 --- a/api/tests/unit_tests/repositories/test_workflow_draft_sync_repository.py +++ b/api/tests/unit_tests/repositories/test_workflow_draft_sync_repository.py @@ -1,4 +1,5 @@ import json +from uuid import uuid4 import pytest import pytest_mock @@ -20,6 +21,7 @@ def test_graphs_stay_opaque_to_lua_and_round_trip_without_changing_values( change = ServerDraftChange( revision="revision-1", hash="hash-1", + client_save_id=str(uuid4()), updated_at=1, graph={"nodes": [{"id": "node", "data": {"items": [], "seed": 9007199254740993}}], "edges": []}, previous_graph=previous, @@ -31,6 +33,7 @@ def test_graphs_stay_opaque_to_lua_and_round_trip_without_changing_values( assert key_count == 1 assert key == "prefix:workflow_server_draft:{app}" encoded = json.loads(payload) + assert encoded["client_save_id"] == change.client_save_id assert isinstance(encoded["graph"], str) assert encoded["previous_graph"] is None or isinstance(encoded["previous_graph"], str) _installed_spy_1 = installed_mock(redis.get) diff --git a/api/tests/unit_tests/services/test_workflow_collaboration_service.py b/api/tests/unit_tests/services/test_workflow_collaboration_service.py index 4414debdc74..4e7fa4aca21 100644 --- a/api/tests/unit_tests/services/test_workflow_collaboration_service.py +++ b/api/tests/unit_tests/services/test_workflow_collaboration_service.py @@ -1,7 +1,9 @@ +import json import logging from collections.abc import Generator, Iterator from contextlib import contextmanager from unittest.mock import MagicMock, Mock, patch +from uuid import uuid4 import pytest from socketio.exceptions import TimeoutError as SocketIOTimeoutError @@ -13,6 +15,7 @@ from extensions.ext_redis import RedisClientWrapper from models.account import Account, Tenant from models.base import TypeBase from models.model import App, AppMode +from models.workflow import Workflow, WorkflowType from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository from services.workflow_collaboration_service import SYNC_REQUEST_TIMEOUT_SECONDS, WorkflowCollaborationService from tests.helpers.unit.config_override import config_overrides_context @@ -33,8 +36,9 @@ def collaboration_redis( @pytest.fixture -def real_service(collaboration_redis: RedisClientWrapper) -> ServiceFixture: +def real_service(collaboration_redis: RedisClientWrapper, db_session: Session) -> ServiceFixture: assert collaboration_redis._require_client() is not None + assert db_session.is_active repository = WorkflowCollaborationRepository() socketio = Mock() return WorkflowCollaborationService(repository, socketio, server_id="server-1"), repository, socketio @@ -44,7 +48,7 @@ def real_service(collaboration_redis: RedisClientWrapper) -> ServiceFixture: def db_session(sqlite_engine: Engine) -> Iterator[Session]: """Provide a real session for tenant-scoped workflow app access checks.""" - TypeBase.metadata.create_all(sqlite_engine, tables=[App.__table__]) + TypeBase.metadata.create_all(sqlite_engine, tables=[App.__table__, Workflow.__table__]) with Session(sqlite_engine, expire_on_commit=False) as session: yield session @@ -62,7 +66,97 @@ def _app(*, app_id: str, tenant_id: str, maintainer: str | None = None) -> App: ) +def _draft(*, tenant_id: str = "t-1", version: str = Workflow.VERSION_DRAFT) -> Workflow: + return Workflow( + app_id="wf-1", + tenant_id=tenant_id, + version=version, + type=WorkflowType.WORKFLOW, + created_by="u-1", + graph=json.dumps({"nodes": [{"id": "start", "data": {"type": "start"}}], "edges": []}), + _features="{}", + draft_client_save_id=str(uuid4()), + ) + + class TestWorkflowCollaborationService: + def test_initial_leader_status_includes_authoritative_draft_receipt( + self, real_service: ServiceFixture, db_session: Session + ) -> None: + collaboration_service, _repository, socketio = real_service + socketio.get_session.return_value = {"user_id": "u-1", "username": "Jane", "tenant_id": "t-1"} + draft = _draft() + db_session.add_all([_app(app_id="wf-1", tenant_id="t-1"), draft]) + db_session.commit() + + with config_overrides_context(RBAC_ENABLED=False): + result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1", session=db_session) + + assert result == ("u-1", True) + assert not db_session.in_transaction() + socketio.emit.assert_any_call( + "status", + { + "isLeader": True, + "draftBaseline": {"hash": draft.unique_hash, "client_save_id": draft.draft_client_save_id}, + }, + room="sid-1", + ) + + def test_promotion_status_refreshes_receipt_after_same_graph_save_without_remote_socket_lookup( + self, real_service: ServiceFixture, db_session: Session + ) -> None: + collaboration_service, repository, socketio = real_service + draft = _draft() + db_session.add_all([_app(app_id="wf-1", tenant_id="t-1"), draft]) + db_session.commit() + initial_hash = draft.unique_hash + initial_receipt = draft.draft_client_save_id + draft.draft_client_save_id = str(uuid4()) + db_session.commit() + assert draft.unique_hash == initial_hash + assert draft.draft_client_save_id != initial_receipt + seed_session(repository, "sid-1") + repository.set_session_info( + "wf-1", + { + "sid": "sid-2", + "user_id": "u-2", + "username": "Remote collaborator", + "avatar": None, + "connected_at": 2, + "server_id": "server-2", + }, + ) + + collaboration_service.broadcast_leader_change("wf-1", "sid-2") + + socketio.get_session.assert_not_called() + socketio.emit.assert_any_call("status", {"isLeader": False}, room="sid-1") + socketio.emit.assert_any_call( + "status", + {"isLeader": True, "draftBaseline": {"hash": initial_hash, "client_save_id": draft.draft_client_save_id}}, + room="sid-2", + ) + + @pytest.mark.parametrize("excluded", ["wrong-tenant", "published"]) + def test_promotion_baseline_excludes_other_owner_versions( + self, real_service: ServiceFixture, db_session: Session, excluded: str + ) -> None: + collaboration_service, repository, socketio = real_service + app = _app(app_id="wf-1", tenant_id="t-1") + draft = _draft( + tenant_id="other-tenant" if excluded == "wrong-tenant" else "t-1", + version="published" if excluded == "published" else Workflow.VERSION_DRAFT, + ) + db_session.add_all([app, draft]) + db_session.commit() + seed_session(repository, "sid-1") + + collaboration_service.broadcast_leader_change("wf-1", "sid-1") + + socketio.emit.assert_called_once_with("status", {"isLeader": True}, room="sid-1") + def test_authorize_and_join_workflow_room_returns_leader_status( self, real_service: ServiceFixture, db_session: Session ) -> None: @@ -386,8 +480,11 @@ class TestWorkflowCollaborationService: assert result == ({"msg": "invalid sync_request"}, 400) socketio.call.assert_not_called() + @pytest.mark.parametrize( + "receipt", [{}, {"clientSaveId": None}, {"clientSaveId": "0bd61160-cf57-46e0-9a83-1cf1dd155a27"}] + ) def test_relay_collaboration_event_sync_request_forwards_to_active_leader( - self, real_service: ServiceFixture + self, real_service: ServiceFixture, receipt: dict[str, object] ) -> None: collaboration_service, repository, socketio = real_service seed_session(repository, "sid-1") @@ -403,7 +500,7 @@ class TestWorkflowCollaborationService: "graph_active": True, }, ) - socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456} + socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456, **receipt} payload = { "type": "sync_request", "data": {"reason": "join", "requestId": "req-1"}, @@ -423,6 +520,7 @@ class TestWorkflowCollaborationService: "success": True, "hash": "hash-2", "updatedAt": 456, + **receipt, }, 200, ) @@ -506,6 +604,7 @@ class TestWorkflowCollaborationService: {"success": "true", "hash": "hash-2", "updatedAt": 456}, {"success": True, "hash": "", "updatedAt": 456}, {"success": True, "hash": "hash-2", "updatedAt": True}, + {"success": True, "hash": "hash-2", "updatedAt": 456, "clientSaveId": 123}, ], ) def test_relay_collaboration_event_sync_request_rejects_invalid_target_response( diff --git a/api/tests/unit_tests/services/test_workflow_draft_sync_service.py b/api/tests/unit_tests/services/test_workflow_draft_sync_service.py index 0386f5f3e88..1e6eb136c5d 100644 --- a/api/tests/unit_tests/services/test_workflow_draft_sync_service.py +++ b/api/tests/unit_tests/services/test_workflow_draft_sync_service.py @@ -2,6 +2,7 @@ import base64 import json from datetime import datetime from unittest.mock import Mock, patch +from uuid import uuid4 import pytest import pytest_mock @@ -29,6 +30,7 @@ def workflow() -> Workflow: created_by="u-1", graph=json.dumps({"nodes": [{"id": "new"}], "edges": []}), _features="{}", + draft_client_save_id=str(uuid4()), ) row.updated_at = datetime(2026, 9, 20, 12, 0, 0) return row @@ -50,6 +52,7 @@ def setup( change = ServerDraftChange( revision="revision-2", hash=workflow.unique_hash, + client_save_id=workflow.draft_client_save_id, updated_at=2, graph=workflow.graph_dict, previous_graph={"nodes": [{"id": "old"}], "edges": []}, @@ -97,8 +100,11 @@ def test_broadcasts_one_accepted_update_including_the_writer_and_replays_it_on_r first, status = service.sync("follower", session=session) assert status == 200 assert first["update"] == b"accepted-update" + assert first["client_save_id"] == change.client_save_id + assert repository.announce.call_args.args[1].client_save_id == change.client_save_id socketio.emit.assert_called_once_with("server_draft_update", first, room="app-1") assert socketio.call.call_args.kwargs["to"] == "leader" + assert socketio.call.call_args.args[1]["client_save_id"] == change.client_save_id assert len(session.info["completed_transactions"]) == 1 assert not session.in_transaction() @@ -205,6 +211,7 @@ def test_notification_carries_a_version_and_preserves_the_patch_baseline(workflo assert change.previous_graph == before assert change.graph == workflow.graph_dict assert change.hash == workflow.unique_hash + assert change.client_save_id == workflow.draft_client_save_id socketio.emit.assert_called_once_with("server_draft_changed", {"revision": change.revision}, room="app-1") diff --git a/packages/contracts/generated/api/console/apps/types.gen.ts b/packages/contracts/generated/api/console/apps/types.gen.ts index fc2bb9c61d9..aa3c1a8fcbd 100644 --- a/packages/contracts/generated/api/console/apps/types.gen.ts +++ b/packages/contracts/generated/api/console/apps/types.gen.ts @@ -1321,6 +1321,7 @@ export type DefaultBlockConfigResponse = { } export type WorkflowResponse = { + client_save_id?: string | null conversation_variables: Array created_at: number created_by?: SimpleAccountResponse | null @@ -1350,6 +1351,8 @@ export type DraftWorkflowNotExistError = ErrorResponse & { export type SyncDraftWorkflowPayload = { _is_collaborative?: boolean + base_client_save_id?: string | null + client_save_id?: string | null conversation_variables?: Array<{ [key: string]: unknown }> @@ -1362,9 +1365,11 @@ export type SyncDraftWorkflowPayload = { [key: string]: unknown } hash?: string | null + predecessor_saves?: Array } export type SyncDraftWorkflowResponse = { + client_save_id?: string | null hash: string result: string updated_at: number @@ -2643,6 +2648,11 @@ export type SyncEnvironmentVariablePatchPayload = { }> } +export type PredecessorDraftSavePayload = { + client_save_id: string + environment_variable_patch?: SyncEnvironmentVariablePatchPayload | null +} + export type ConversationVariableItemPayload = { description?: string | null id?: string | null diff --git a/packages/contracts/generated/api/console/apps/zod.gen.ts b/packages/contracts/generated/api/console/apps/zod.gen.ts index 75d0b584e59..0b0733355f6 100644 --- a/packages/contracts/generated/api/console/apps/zod.gen.ts +++ b/packages/contracts/generated/api/console/apps/zod.gen.ts @@ -1373,6 +1373,7 @@ export const zDraftWorkflowNotExistError = zErrorResponse.and( * SyncDraftWorkflowResponse */ export const zSyncDraftWorkflowResponse = z.object({ + client_save_id: z.uuid().nullish(), hash: z.string(), result: z.string(), updated_at: z.int(), @@ -2832,6 +2833,7 @@ export const zPipelineVariableResponse = z.object({ * WorkflowResponse */ export const zWorkflowResponse = z.object({ + client_save_id: z.uuid().nullish(), conversation_variables: z.array(zWorkflowConversationVariableResponse), created_at: z.int(), created_by: zSimpleAccountResponse.nullish(), @@ -2868,17 +2870,28 @@ export const zSyncEnvironmentVariablePatchPayload = z.object({ environment_variables: z.array(z.record(z.string(), z.unknown())).optional(), }) +/** + * PredecessorDraftSavePayload + */ +export const zPredecessorDraftSavePayload = z.object({ + client_save_id: z.uuid(), + environment_variable_patch: zSyncEnvironmentVariablePatchPayload.nullish(), +}) + /** * SyncDraftWorkflowPayload */ export const zSyncDraftWorkflowPayload = z.object({ _is_collaborative: z.boolean().optional().default(false), + base_client_save_id: z.uuid().nullish(), + client_save_id: z.uuid().nullish(), conversation_variables: z.array(z.record(z.string(), z.unknown())).optional(), environment_variable_patch: zSyncEnvironmentVariablePatchPayload.nullish(), features: z.record(z.string(), z.unknown()), force: z.boolean().optional().default(false), graph: z.record(z.string(), z.unknown()), hash: z.string().nullish(), + predecessor_saves: z.array(zPredecessorDraftSavePayload).optional(), }) /** diff --git a/packages/contracts/generated/api/console/rag/types.gen.ts b/packages/contracts/generated/api/console/rag/types.gen.ts index 49341522373..509cb4c6ca7 100644 --- a/packages/contracts/generated/api/console/rag/types.gen.ts +++ b/packages/contracts/generated/api/console/rag/types.gen.ts @@ -195,6 +195,7 @@ export type DefaultBlockConfigResponse = { } export type WorkflowResponse = { + client_save_id?: string | null conversation_variables: Array created_at: number created_by?: SimpleAccountResponse | null diff --git a/packages/contracts/generated/api/console/rag/zod.gen.ts b/packages/contracts/generated/api/console/rag/zod.gen.ts index e4ce174326b..edbf91fbe82 100644 --- a/packages/contracts/generated/api/console/rag/zod.gen.ts +++ b/packages/contracts/generated/api/console/rag/zod.gen.ts @@ -529,6 +529,7 @@ export const zPipelineVariableResponse = z.object({ * WorkflowResponse */ export const zWorkflowResponse = z.object({ + client_save_id: z.uuid().nullish(), conversation_variables: z.array(zWorkflowConversationVariableResponse), created_at: z.int(), created_by: zSimpleAccountResponse.nullish(), diff --git a/packages/contracts/generated/api/console/snippets/types.gen.ts b/packages/contracts/generated/api/console/snippets/types.gen.ts index 9d8d655e070..feb33f7006f 100644 --- a/packages/contracts/generated/api/console/snippets/types.gen.ts +++ b/packages/contracts/generated/api/console/snippets/types.gen.ts @@ -57,6 +57,7 @@ export type SnippetWorkflowPaginationResponse = { export type DefaultBlockConfigsResponse = Array export type SnippetWorkflowResponse = { + client_save_id?: string | null conversation_variables: Array created_at: number created_by?: SimpleAccountResponse | null diff --git a/packages/contracts/generated/api/console/snippets/zod.gen.ts b/packages/contracts/generated/api/console/snippets/zod.gen.ts index ee89768d544..96d1c74f579 100644 --- a/packages/contracts/generated/api/console/snippets/zod.gen.ts +++ b/packages/contracts/generated/api/console/snippets/zod.gen.ts @@ -407,6 +407,7 @@ export const zPipelineVariableResponse = z.object({ * SnippetWorkflowResponse */ export const zSnippetWorkflowResponse = z.object({ + client_save_id: z.uuid().nullish(), conversation_variables: z.array(zWorkflowConversationVariableResponse), created_at: z.int(), created_by: zSimpleAccountResponse.nullish(), diff --git a/web/app/components/workflow-app/components/__tests__/workflow-main.creation.spec.tsx b/web/app/components/workflow-app/components/__tests__/workflow-main.creation.spec.tsx index c47f2dcf18d..24859fe6b5a 100644 --- a/web/app/components/workflow-app/components/__tests__/workflow-main.creation.spec.tsx +++ b/web/app/components/workflow-app/components/__tests__/workflow-main.creation.spec.tsx @@ -92,6 +92,7 @@ vi.mock('@/app/components/workflow/collaboration/core/collaboration-manager', () createServerDraftUpdate: vi.fn(), onServerDraftRequest: () => () => {}, onServerDraftApplied: () => () => {}, + onDraftBaseline: () => () => {}, onGraphReadyChange: (listener: (ready: boolean) => void) => { mocks.graphReadyListener = listener // Before the collaboration connection starts, graph editing is still allowed. diff --git a/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx b/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx index f5d3284644b..9e5f7173d57 100644 --- a/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx +++ b/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx @@ -90,6 +90,7 @@ const events = { cursors: new Set[0]>(), ready: new Set[0]>(), applied: new Set[0]>(), + baseline: new Set[0]>(), } const initialNodes = [ createStartNode({ id: 'start', data: { title: 'Workflow entry', variables: [] } }), @@ -146,6 +147,12 @@ beforeEach(() => { events.applied.delete(handler) } }) + vi.spyOn(collaborationManager, 'onDraftBaseline').mockImplementation((handler) => { + events.baseline.add(handler) + return () => { + events.baseline.delete(handler) + } + }) vi.spyOn(collaborationManager, 'canPersistLocalGraph').mockReturnValue(true) vi.spyOn(collaborationManager, 'refreshGraphSynchronously').mockImplementation(() => {}) vi.spyOn(collaborationManager, 'replaceGraphFromReactFlow').mockReturnValue(true) @@ -424,6 +431,166 @@ it('saves the trusted graph on a directed request and acknowledges the persisted expect(collaborationManager.refreshGraphSynchronously).toHaveBeenCalled() }) +it('uses the latest save receipt after a trusted follower becomes the saver without a graph change', async () => { + vi.mocked(collaborationManager.getIsLeader).mockReturnValue(false) + api.saveDraft.mockImplementationOnce(async ({ body }) => ({ + result: 'success', + hash: 'shared-hash', + updated_at: 3, + client_save_id: body.client_save_id, + })) + const latestClientSaveId = '00000000-0000-4000-8000-000000000002' + const { store } = await renderMain({ + initialStoreState: { + syncWorkflowDraftHash: 'shared-hash', + syncWorkflowDraftSaveId: '00000000-0000-4000-8000-000000000001', + syncWorkflowDraftIsEmpty: false, + }, + }) + act(() => { + vi.mocked(collaborationManager.getIsLeader).mockReturnValue(true) + events.baseline.forEach((handler) => + handler({ hash: 'shared-hash', client_save_id: latestClientSaveId }), + ) + }) + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'shared-hash', + syncWorkflowDraftSaveId: latestClientSaveId, + syncWorkflowDraftIsEmpty: false, + lastSavedWorkflowDraft: null, + }) + const acknowledge = vi.fn() + act(() => { + events.sync.forEach((handler) => handler({ requestId: 'promoted-save', acknowledge })) + }) + await waitFor(() => expect(acknowledge).toHaveBeenCalledOnce()) + + expect(api.saveDraft.mock.lastCall?.[0].body).toMatchObject({ + hash: 'shared-hash', + base_client_save_id: latestClientSaveId, + graph: { nodes: [expect.objectContaining({ id: 'start' })] }, + }) + expect(acknowledge).toHaveBeenCalledWith({ + success: true, + hash: 'shared-hash', + updatedAt: 3, + clientSaveId: api.saveDraft.mock.lastCall?.[0].body.client_save_id, + }) +}) + +it('does not borrow a promoted saver baseline for a different graph', async () => { + const localClientSaveId = '00000000-0000-4000-8000-000000000001' + const { store } = await renderMain({ + initialStoreState: { + syncWorkflowDraftHash: 'local-hash', + syncWorkflowDraftSaveId: localClientSaveId, + }, + }) + act(() => { + events.baseline.forEach((handler) => + handler({ + hash: 'different-server-hash', + client_save_id: '00000000-0000-4000-8000-000000000002', + }), + ) + }) + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'local-hash', + syncWorkflowDraftSaveId: localClientSaveId, + lastSavedWorkflowDraft: null, + }) + const acknowledge = vi.fn() + act(() => { + events.sync.forEach((handler) => handler({ requestId: 'old-graph-save', acknowledge })) + }) + await waitFor(() => expect(acknowledge).toHaveBeenCalledOnce()) + expect(api.saveDraft.mock.lastCall?.[0].body).toMatchObject({ + hash: 'local-hash', + base_client_save_id: localClientSaveId, + }) +}) + +it('keeps the in-flight local save baseline when a promotion receipt arrives', async () => { + const pending = createDeferred>>() + api.saveDraft.mockReturnValueOnce(pending.promise) + const { store } = await renderMain({ + initialStoreState: { + syncWorkflowDraftHash: 'shared-hash', + syncWorkflowDraftSaveId: '00000000-0000-4000-8000-000000000001', + }, + }) + const acknowledge = vi.fn() + act(() => { + events.sync.forEach((handler) => handler({ requestId: 'old-save', acknowledge })) + }) + await waitFor(() => expect(api.saveDraft).toHaveBeenCalledOnce()) + const latestClientSaveId = '00000000-0000-4000-8000-000000000002' + await act(async () => { + events.baseline.forEach((handler) => + handler({ hash: 'shared-hash', client_save_id: latestClientSaveId }), + ) + expect(store.getState().syncWorkflowDraftSaveId).toBe('00000000-0000-4000-8000-000000000001') + pending.resolve({ + result: 'success', + hash: 'local-save-hash', + updated_at: 2, + client_save_id: api.saveDraft.mock.lastCall?.[0].body.client_save_id, + }) + }) + const localClientSaveId = api.saveDraft.mock.lastCall?.[0].body.client_save_id + await waitFor(() => + expect(acknowledge).toHaveBeenCalledWith({ + success: true, + hash: 'local-save-hash', + updatedAt: 2, + clientSaveId: localClientSaveId, + }), + ) + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'local-save-hash', + syncWorkflowDraftSaveId: localClientSaveId, + hasWorkflowDraftConflict: false, + }) +}) + +it('keeps an unacknowledged local save baseline when a promotion receipt arrives', async () => { + api.saveDraft.mockRejectedValueOnce(new TypeError('Failed to fetch')) + const originalClientSaveId = '00000000-0000-4000-8000-000000000001' + const { store } = await renderMain({ + initialStoreState: { + syncWorkflowDraftHash: 'shared-hash', + syncWorkflowDraftSaveId: originalClientSaveId, + }, + }) + const acknowledge = vi.fn() + act(() => { + events.sync.forEach((handler) => handler({ requestId: 'unacknowledged-save', acknowledge })) + }) + await waitFor(() => expect(acknowledge).toHaveBeenCalledWith({ success: false })) + const unacknowledgedClientSaveId = api.saveDraft.mock.lastCall?.[0].body.client_save_id + act(() => { + events.baseline.forEach((handler) => + handler({ + hash: 'shared-hash', + client_save_id: '00000000-0000-4000-8000-000000000002', + }), + ) + }) + expect(store.getState().syncWorkflowDraftSaveId).toBe(originalClientSaveId) + const retryAcknowledge = vi.fn() + act(() => { + events.sync.forEach((handler) => + handler({ requestId: 'retry-save', acknowledge: retryAcknowledge }), + ) + }) + await waitFor(() => expect(retryAcknowledge).toHaveBeenCalledOnce()) + expect(api.saveDraft.mock.lastCall?.[0].body).toMatchObject({ + hash: 'shared-hash', + base_client_save_id: originalClientSaveId, + predecessor_saves: [{ client_save_id: unacknowledgedClientSaveId }], + }) +}) + it('replaces the canvas from a workflow update and records its revision', async () => { const { store } = await renderMain({ initialStoreState: { syncWorkflowDraftIsEmpty: true } }) api.fetchDraft.mockResolvedValue( diff --git a/web/app/components/workflow-app/components/workflow-main.tsx b/web/app/components/workflow-app/components/workflow-main.tsx index e41950845d3..8439e6ede92 100644 --- a/web/app/components/workflow-app/components/workflow-main.tsx +++ b/web/app/components/workflow-app/components/workflow-main.tsx @@ -322,6 +322,11 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => { useEffect(() => { if (!appId || !isCollaborationEnabled) return + const unsubscribeBaseline = collaborationManager.onDraftBaseline((baseline) => { + workflowStore + .getState() + .adoptWorkflowDraftSaveId(appId, baseline.hash, baseline.client_save_id) + }) const unsubscribeRequest = collaborationManager.onServerDraftRequest( ({ change, acknowledge }) => { const localStartPlaceholderNodes = collaborationManager @@ -349,7 +354,7 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => { // A refresh already invalidated queued saves. Keep that refresh alive // while its server-approved graph and matching metadata arrive. if (!state.isSyncingWorkflowDraft) state.invalidateWorkflowDraftSync() - state.setSyncWorkflowDraftHash(update.hash) + state.setSyncWorkflowDraftHash(update.hash, undefined, update.client_save_id) state.setDraftUpdatedAt(update.updated_at) if (hadUnsavedChanges && !state.isSyncingWorkflowDraft) { state.markWorkflowDraftDirty() @@ -358,6 +363,7 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => { } }) return () => { + unsubscribeBaseline() unsubscribeRequest() unsubscribeApplied() } @@ -404,7 +410,7 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => { .then((result) => { acknowledge( result - ? { success: true, hash: result.hash, updatedAt: result.updatedAt } + ? { success: true, ...result } : { success: false, ...(workflowStore.getState().hasWorkflowDraftConflict diff --git a/web/app/components/workflow-app/hooks/__tests__/use-nodes-sync-draft.spec.ts b/web/app/components/workflow-app/hooks/__tests__/use-nodes-sync-draft.spec.ts index 173b1874582..22cbfe13550 100644 --- a/web/app/components/workflow-app/hooks/__tests__/use-nodes-sync-draft.spec.ts +++ b/web/app/components/workflow-app/hooks/__tests__/use-nodes-sync-draft.spec.ts @@ -1,4 +1,9 @@ -import type { AppMode, WorkflowResponse } from '@dify/contracts/api/console/apps/types.gen' +import type { + AppMode, + SyncDraftWorkflowPayload, + SyncDraftWorkflowResponse, + WorkflowResponse, +} from '@dify/contracts/api/console/apps/types.gen' import type { Features } from '@/app/components/base/features/types' import type { Shape } from '@/app/components/workflow/store/workflow' import type { Edge, Node } from '@/app/components/workflow/types' @@ -44,7 +49,13 @@ vi.mock('@/app/components/workflow/collaboration/core/collaboration-manager', () }, })) -const mockSyncWorkflowDraft = vi.fn() +const mockSyncWorkflowDraft = + vi.fn< + (request: { + url: string + params: SyncDraftWorkflowPayload + }) => Promise> + >() const mockFetchWorkflowDraft = vi.fn() // Stand in for the draft sync request at the client boundary. Calls are recorded as the // request they send (endpoint and body), and rejections pass through the console link's @@ -61,7 +72,10 @@ vi.mock('@/service/console', async (importOriginal) => { path: ['apps', 'byAppId', 'workflows', 'draft', 'get'], next: () => mockFetchWorkflowDraft(input), }) - const post = (input: { params: { app_id: string }; body: unknown }, options?: unknown) => { + const post = ( + input: { params: { app_id: string }; body: SyncDraftWorkflowPayload }, + options?: unknown, + ) => { mockDraftRequestContext(options) return validate({ context: {}, @@ -197,6 +211,22 @@ describe('useNodesSyncDraft', () => { }) describe('payload', () => { + it('saves drafts when randomUUID is unavailable in the hosting context', async () => { + vi.stubGlobal('crypto', { getRandomValues: crypto.getRandomValues.bind(crypto) }) + try { + const { result } = renderUseNodesSyncDraft() + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce() + expect(mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/, + ) + } finally { + vi.unstubAllGlobals() + } + }) + it.each(['draft sync', 'page close'] as const)( 'removes Workflow LLM memory from %s without changing the canvas nodes', async (savePath) => { @@ -355,6 +385,7 @@ describe('useNodesSyncDraft', () => { }, conversation_variables: conversationVariables, hash: 'latest-hash', + client_save_id: expect.any(String), }), ) expect(store.getState()).toMatchObject({ @@ -895,6 +926,75 @@ describe('useNodesSyncDraft', () => { }) describe('failures and skipped saves', () => { + it.each([400, 408, 429])( + 'preserves earlier uncertain patches while handling a later HTTP %s rejection', + async (status) => { + mockSyncWorkflowDraft + .mockRejectedValueOnce(new TypeError('Failed to fetch')) + .mockRejectedValueOnce( + createHttpError(new Response(null, { status }), { code: 'save_rejected', status }), + ) + const earlierVariables: EnvironmentVariablePatch['environmentVariables'] = [ + { + id: 'earlier', + name: 'earlier', + value_type: 'string', + value: 'Earlier edit', + description: '', + }, + ] + const latestVariables: EnvironmentVariablePatch['environmentVariables'] = [ + { + id: 'latest', + name: 'latest', + value_type: 'string', + value: 'Latest edit', + description: '', + }, + ] + const { result, store } = renderUseNodesSyncDraft() + await act(async () => { + await result.current.doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: earlierVariables, + deletedEnvironmentVariableIds: [], + }, + }) + await result.current.doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: latestVariables, + deletedEnvironmentVariableIds: [], + }, + }) + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(3) + const retry = mockSyncWorkflowDraft.mock.calls[2]?.[0].params + expect(retry?.predecessor_saves).toEqual([ + { + client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + environment_variable_patch: { + environment_variables: earlierVariables, + deleted_environment_variable_ids: [], + }, + }, + { + client_save_id: mockSyncWorkflowDraft.mock.calls[1]?.[0].params.client_save_id, + ...(status === 400 + ? {} + : { + environment_variable_patch: { + environment_variables: latestVariables, + deleted_environment_variable_ids: [], + }, + }), + }, + ]) + expect(retry?.environment_variable_patch).toBeUndefined() + expect(store.getState().hasWorkflowDraftConflict).toBe(false) + }, + ) + it.each([ { scenario: 'an out-of-date draft', @@ -1241,8 +1341,14 @@ describe('useNodesSyncDraft', () => { }, ) - it('waits for the leader save result when the current user is a follower', async () => { + it('adopts the leader receipt before this follower saves locally', async () => { useCollaborationSession({ connected: true, leader: false }) + const leaderClientSaveId = '00000000-0000-4000-8000-000000000001' + mockCollaborationRequestWorkflowSync.mockResolvedValue({ + hash: 'remote-hash', + updatedAt: 2, + clientSaveId: leaderClientSaveId, + }) const callbacks = { onSuccess: vi.fn(), onError: vi.fn(), onSettled: vi.fn() } const { result, store } = renderUseNodesSyncDraft({ isCollaborationEnabled: true }) @@ -1254,11 +1360,24 @@ describe('useNodesSyncDraft', () => { expect(mockSyncWorkflowDraft).not.toHaveBeenCalled() expect(store.getState()).toMatchObject({ syncWorkflowDraftHash: 'remote-hash', + syncWorkflowDraftSaveId: leaderClientSaveId, draftUpdatedAt: 2000, }) expect(callbacks.onSuccess).toHaveBeenCalledOnce() expect(callbacks.onError).not.toHaveBeenCalled() expect(callbacks.onSettled).toHaveBeenCalledOnce() + useCollaborationSession({ connected: true, leader: true }) + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledWith( + draftRequest( + expect.objectContaining({ + hash: 'remote-hash', + base_client_save_id: leaderClientSaveId, + }), + ), + ) }) it('reports a failed leader save to the follower caller', async () => { @@ -1279,10 +1398,22 @@ describe('useNodesSyncDraft', () => { it('saves a directed sync request locally even before leader status arrives', async () => { useCollaborationSession({ connected: true, leader: false }) + mockSyncWorkflowDraft.mockImplementationOnce(async ({ params }) => ({ + hash: 'directed-save', + updated_at: 1, + client_save_id: params.client_save_id, + })) const { result } = renderUseNodesSyncDraft({ isCollaborationEnabled: true }) await act(async () => { - await result.current.doSyncWorkflowDraft(false, undefined, { forceLocal: true }) + const saved = await result.current.doSyncWorkflowDraft(false, undefined, { + forceLocal: true, + }) + expect(saved).toEqual({ + hash: 'directed-save', + updatedAt: 1, + clientSaveId: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + }) }) expect(mockCollaborationRequestWorkflowSync).not.toHaveBeenCalled() @@ -1372,7 +1503,7 @@ describe('useNodesSyncDraft', () => { it.each(['success', 'conflict'] as const)( 'ignores a stale %s response after a newer draft reload', async (outcome) => { - const pending = createDeferred<{ hash: string; updated_at: number }>() + const pending = createDeferred>() mockSyncWorkflowDraft.mockReturnValueOnce(pending.promise) const { result, store } = renderUseNodesSyncDraft() const onSuccess = vi.fn() @@ -1382,11 +1513,17 @@ describe('useNodesSyncDraft', () => { store.getState().invalidateWorkflowDraftSync() store.getState().setSyncWorkflowDraftHash('reloaded-hash') await act(async () => { - if (outcome === 'success') pending.resolve({ hash: 'old-save', updated_at: 1 }) + if (outcome === 'success') + pending.resolve({ + hash: 'old-save', + updated_at: 1, + client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + }) else pending.reject(draftNotSyncError()) expect(await saving).toBeNull() }) expect(store.getState().syncWorkflowDraftHash).toBe('reloaded-hash') + expect(store.getState().syncWorkflowDraftSaveId).toBeNull() expect(store.getState().hasWorkflowDraftConflict).toBe(false) expect(onSuccess).not.toHaveBeenCalled() expect(onError).not.toHaveBeenCalled() @@ -1396,19 +1533,35 @@ describe('useNodesSyncDraft', () => { it.each(['generation', 'hash'] as const)( 'does not lend an external %s to a queued old graph', async (change) => { - const { result, store } = renderUseNodesSyncDraft() + const { result, store } = renderUseNodesSyncDraft({ + initialStoreState: { syncWorkflowDraftSaveId: '00000000-0000-4000-8000-000000000001' }, + }) const saving = result.current.doSyncWorkflowDraft() if (change === 'generation') store.getState().invalidateWorkflowDraftSync() store.getState().setSyncWorkflowDraftHash('external-hash') expect(await saving).toBeNull() expect(mockSyncWorkflowDraft).not.toHaveBeenCalled() + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce() + expect(mockSyncWorkflowDraft.mock.calls[0]?.[0].params.hash).toBe('external-hash') + expect(mockSyncWorkflowDraft.mock.calls[0]?.[0].params.base_client_save_id).toBeUndefined() }, ) - it('shares successful queued saves and hashes across hook instances', async () => { + it('shares successful queued saves, hashes, and receipts across hook instances', async () => { mockSyncWorkflowDraft - .mockResolvedValueOnce({ hash: 'saved-1', updated_at: 1 }) - .mockResolvedValueOnce({ hash: 'saved-2', updated_at: 2 }) + .mockImplementationOnce(async ({ params }) => ({ + hash: 'saved-1', + updated_at: 1, + client_save_id: params.client_save_id, + })) + .mockImplementationOnce(async ({ params }) => ({ + hash: 'saved-2', + updated_at: 2, + client_save_id: params.client_save_id, + })) const { result, store, setCanvas } = renderSyncDraft( () => [useNodesSyncDraft(), useNodesSyncDraft()] as const, ) @@ -1423,11 +1576,15 @@ describe('useNodesSyncDraft', () => { expect.objectContaining({ params: expect.objectContaining({ hash: 'saved-1', + base_client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, graph: expect.objectContaining({ nodes: [expect.objectContaining({ id: 'latest' })] }), }), }), ) expect(store.getState().syncWorkflowDraftHash).toBe('saved-2') + expect(store.getState().syncWorkflowDraftSaveId).toBe( + mockSyncWorkflowDraft.mock.calls[1]?.[0].params.client_save_id, + ) }) it('blocks normal and keepalive saves while a draft is being applied', async () => { @@ -1453,6 +1610,41 @@ describe('useNodesSyncDraft', () => { }) describe('save and refresh ordering', () => { + it('uses a refreshed hash and receipt instead of an earlier save receipt', async () => { + const refreshedClientSaveId = '00000000-0000-4000-8000-000000000002' + mockSyncWorkflowDraft.mockImplementationOnce(async ({ params }) => ({ + hash: 'saved-locally', + updated_at: 1, + client_save_id: params.client_save_id, + })) + const { result, store } = renderUseNodesSyncDraft() + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + expect(store.getState().syncWorkflowDraftSaveId).toBe( + mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + ) + await act(async () => { + store.getState().invalidateWorkflowDraftSync() + store + .getState() + .setSyncWorkflowDraftHash( + 'saved-remotely', + { nodes: [startNode], edges: [] }, + refreshedClientSaveId, + ) + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenLastCalledWith( + draftRequest( + expect.objectContaining({ + hash: 'saved-remotely', + base_client_save_id: refreshedClientSaveId, + }), + ), + ) + }) + it('preserves an accepted variable patch queued after a background refresh request', async () => { const pending = createDeferred<{ hash: string; updated_at: number }>() mockSyncWorkflowDraft.mockReturnValueOnce(pending.promise) @@ -1525,9 +1717,10 @@ describe('useNodesSyncDraft', () => { it('drains an accepted keepalive save before starting a background full read', async () => { const pending = createDeferred<{ hash: string; updated_at: number }>() + const unloading = createDeferred<{ hash: string; updated_at: number }>() mockSyncWorkflowDraft .mockReturnValueOnce(pending.promise) - .mockResolvedValueOnce({ hash: 'latest-saved-hash', updated_at: 3 }) + .mockReturnValueOnce(unloading.promise) mockFetchDraft.mockResolvedValue({ hash: 'latest-saved-hash', graph: { nodes: [startNode], edges: [] }, @@ -1542,6 +1735,7 @@ describe('useNodesSyncDraft', () => { setCanvas({ nodes: [createNode({ id: 'latest-local-edit' })] }) store.getState().markWorkflowDraftDirty() result.current.save.syncWorkflowDraftWhenPageClose() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) const reading = result.current.refresh.handleRefreshWorkflowDraft(false, { background: true, preserveViewport: true, @@ -1550,13 +1744,20 @@ describe('useNodesSyncDraft', () => { await act(async () => { pending.resolve({ hash: 'first-saved-hash', updated_at: 2 }) await saving + }) + expect(mockFetchDraft).not.toHaveBeenCalled() + await act(async () => { + unloading.resolve({ hash: 'latest-saved-hash', updated_at: 3 }) await reading }) expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) expect(mockSyncWorkflowDraft).toHaveBeenLastCalledWith( expect.objectContaining({ params: expect.objectContaining({ - hash: 'first-saved-hash', + hash: 'hash-123', + predecessor_saves: [ + { client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id }, + ], graph: expect.objectContaining({ nodes: [expect.objectContaining({ id: 'latest-local-edit' })], }), @@ -1571,24 +1772,25 @@ describe('useNodesSyncDraft', () => { expect(store.getState().hasWorkflowDraftConflict).toBe(false) }) - it('queues a keepalive snapshot behind the current save and uses its acknowledged hash', async () => { + it('dispatches the final keepalive snapshot before the pending save completes or the hook unmounts', async () => { const pending = createDeferred<{ hash: string; updated_at: number }>() mockSyncWorkflowDraft.mockReturnValueOnce(pending.promise) - const { result, setCanvas } = renderUseNodesSyncDraft() + const { result, setCanvas, unmount } = renderUseNodesSyncDraft() const saving = result.current.doSyncWorkflowDraft() await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) setCanvas({ nodes: [createNode({ id: 'latest-local-edit' })] }) - result.current.syncWorkflowDraftWhenPageClose() - expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce() - await act(async () => { - pending.resolve({ hash: 'first-saved-hash', updated_at: 2 }) - await saving + act(() => { + result.current.syncWorkflowDraftWhenPageClose() }) - await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2)) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) expect(mockSyncWorkflowDraft).toHaveBeenLastCalledWith( expect.objectContaining({ params: expect.objectContaining({ - hash: 'first-saved-hash', + hash: 'hash-123', + client_save_id: expect.any(String), + predecessor_saves: [ + { client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id }, + ], graph: expect.objectContaining({ nodes: [expect.objectContaining({ id: 'latest-local-edit' })], }), @@ -1598,6 +1800,322 @@ describe('useNodesSyncDraft', () => { expect(mockDraftRequestContext).toHaveBeenLastCalledWith({ context: { silent: true, keepalive: true }, }) + unmount() + await act(async () => { + pending.resolve({ hash: 'first-saved-hash', updated_at: 2 }) + await saving + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + }) + + it.each([ + { firstResponse: 'autosave', autosaveOutcome: 'success' }, + { firstResponse: 'unload', autosaveOutcome: 'success' }, + { firstResponse: 'autosave', autosaveOutcome: 'conflict' }, + { firstResponse: 'unload', autosaveOutcome: 'conflict' }, + ] as const)( + 'keeps the final save when $firstResponse responds first and autosave returns $autosaveOutcome', + async ({ firstResponse, autosaveOutcome }) => { + const autosave = createDeferred<{ hash: string; updated_at: number }>() + const unload = createDeferred<{ hash: string; updated_at: number }>() + mockSyncWorkflowDraft + .mockReturnValueOnce(autosave.promise) + .mockReturnValueOnce(unload.promise) + const callbacks = { onError: vi.fn(), onSettled: vi.fn() } + const { result, store, setCanvas } = renderSyncDraft( + () => [useNodesSyncDraft(), useNodesSyncDraft()] as const, + ) + const saving = result.current[0].doSyncWorkflowDraft(false, callbacks) + await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) + setCanvas({ nodes: [createNode({ id: 'final-edit' })] }) + result.current[1].syncWorkflowDraftWhenPageClose() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + + const finishAutosave = () => { + if (autosaveOutcome === 'success') autosave.resolve({ hash: 'older-save', updated_at: 2 }) + else autosave.reject(draftNotSyncError()) + } + if (firstResponse === 'autosave') { + await act(async () => { + finishAutosave() + await saving + }) + expect(store.getState().hasWorkflowDraftConflict).toBe(false) + } + await act(async () => { + unload.resolve({ hash: 'final-save', updated_at: 3 }) + }) + await waitFor(() => expect(store.getState().syncWorkflowDraftHash).toBe('final-save')) + if (firstResponse === 'unload') { + await act(async () => { + finishAutosave() + await saving + }) + } + + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'final-save', + draftUpdatedAt: 3000, + hasWorkflowDraftConflict: false, + }) + expect(callbacks.onError).not.toHaveBeenCalled() + expect(callbacks.onSettled).toHaveBeenCalledOnce() + }, + ) + + it('keeps a real conflict from the final keepalive save despite an older successful response', async () => { + const autosave = createDeferred<{ hash: string; updated_at: number }>() + const unload = createDeferred<{ hash: string; updated_at: number }>() + mockSyncWorkflowDraft + .mockReturnValueOnce(autosave.promise) + .mockReturnValueOnce(unload.promise) + const { result, store, setCanvas } = renderUseNodesSyncDraft() + const saving = result.current.doSyncWorkflowDraft() + await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) + setCanvas({ nodes: [createNode({ id: 'unsaved-local-edit' })] }) + result.current.syncWorkflowDraftWhenPageClose() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + + await act(async () => { + unload.reject(draftNotSyncError()) + }) + await waitFor(() => expect(store.getState().hasWorkflowDraftConflict).toBe(true)) + await act(async () => { + autosave.resolve({ hash: 'older-save', updated_at: 2 }) + await saving + }) + + expect(store.getState().hasWorkflowDraftConflict).toBe(true) + expect(await result.current.doSyncWorkflowDraft()).toBeNull() + expect(mockFetchDraft).not.toHaveBeenCalled() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + }) + + it('includes pending variable patches in the final snapshot without replaying queued older graphs', async () => { + const pending = createDeferred<{ hash: string; updated_at: number }>() + const unloading = createDeferred<{ hash: string; updated_at: number }>() + mockSyncWorkflowDraft + .mockReturnValueOnce(pending.promise) + .mockReturnValueOnce(unloading.promise) + const firstVariables: EnvironmentVariablePatch['environmentVariables'] = [ + { id: 'edited', name: 'edited', value_type: 'string', value: 'old', description: '' }, + { id: 'removed', name: 'removed', value_type: 'string', value: 'old', description: '' }, + ] + const latestVariables: EnvironmentVariablePatch['environmentVariables'] = [ + { id: 'edited', name: 'edited', value_type: 'string', value: 'latest', description: '' }, + { + id: 'restored', + name: 'restored', + value_type: 'string', + value: 'restored', + description: '', + }, + ] + const { result, setCanvas, store } = renderSyncDraft( + () => [useNodesSyncDraft(), useNodesSyncDraft()] as const, + ) + const saving = result.current[0].doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: firstVariables, + deletedEnvironmentVariableIds: ['restored'], + }, + }) + await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) + setCanvas({ nodes: [createNode({ id: 'queued-old-graph' })] }) + const queued = result.current[0].doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: latestVariables, + deletedEnvironmentVariableIds: ['removed'], + }, + }) + setCanvas({ nodes: [createNode({ id: 'final-edit' })] }) + result.current[1].syncWorkflowDraftWhenPageClose() + + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + expect(mockSyncWorkflowDraft).toHaveBeenLastCalledWith( + draftRequest( + expect.objectContaining({ + hash: 'hash-123', + predecessor_saves: [ + { + client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + environment_variable_patch: { + environment_variables: firstVariables, + deleted_environment_variable_ids: ['restored'], + }, + }, + { + client_save_id: expect.any(String), + environment_variable_patch: { + environment_variables: latestVariables, + deleted_environment_variable_ids: ['removed'], + }, + }, + ], + graph: expect.objectContaining({ + nodes: [expect.objectContaining({ id: 'final-edit' })], + }), + }), + ), + ) + expect( + mockSyncWorkflowDraft.mock.calls[1]?.[0].params.environment_variable_patch, + ).toBeUndefined() + await act(async () => { + unloading.resolve({ hash: 'final-save', updated_at: 3 }) + pending.reject(draftNotSyncError()) + await Promise.all([saving, queued]) + await store.getState().enqueueWorkflowDraftOperation(async () => undefined) + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + expect(store.getState().syncWorkflowDraftHash).toBe('final-save') + expect(store.getState().hasWorkflowDraftConflict).toBe(false) + }) + + it('retains each predecessor patch once across repeated unload saves', async () => { + const pending = createDeferred>() + const unloading = createDeferred>() + const finalUnload = createDeferred>() + mockSyncWorkflowDraft + .mockReturnValueOnce(pending.promise) + .mockReturnValueOnce(unloading.promise) + .mockReturnValueOnce(finalUnload.promise) + const variables: EnvironmentVariablePatch['environmentVariables'] = [ + { id: 'env-1', name: 'topic', value_type: 'string', value: 'Local edit', description: '' }, + ] + const { result, store, setCanvas } = renderUseNodesSyncDraft() + const saving = result.current.doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: variables, + deletedEnvironmentVariableIds: ['removed'], + }, + }) + await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) + const firstClientSaveId = mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id + result.current.syncWorkflowDraftWhenPageClose() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + const unloadingClientSaveId = mockSyncWorkflowDraft.mock.calls[1]?.[0].params.client_save_id + await act(async () => { + pending.resolve({ hash: 'first-save', updated_at: 1, client_save_id: firstClientSaveId }) + await saving + }) + setCanvas({ nodes: [createNode({ id: 'final-edit' })] }) + result.current.syncWorkflowDraftWhenPageClose() + + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(3) + const finalRequest = mockSyncWorkflowDraft.mock.calls[2]?.[0].params + expect(finalRequest).toMatchObject({ + hash: 'hash-123', + predecessor_saves: [ + { + client_save_id: firstClientSaveId, + environment_variable_patch: { + environment_variables: variables, + deleted_environment_variable_ids: ['removed'], + }, + }, + { client_save_id: unloadingClientSaveId }, + ], + graph: expect.objectContaining({ nodes: [expect.objectContaining({ id: 'final-edit' })] }), + }) + expect(finalRequest?.environment_variable_patch).toBeUndefined() + expect(finalRequest?.predecessor_saves?.[1]?.environment_variable_patch).toBeUndefined() + await act(async () => { + finalUnload.resolve({ + hash: 'final-save', + updated_at: 3, + client_save_id: finalRequest?.client_save_id, + }) + unloading.reject(draftNotSyncError()) + await store.getState().enqueueWorkflowDraftOperation(async () => undefined) + }) + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'final-save', + syncWorkflowDraftSaveId: finalRequest?.client_save_id, + hasWorkflowDraftConflict: false, + }) + }) + + it('retries queued variable changes with the latest graph after an unload transport failure', async () => { + const pending = createDeferred>() + const unloading = createDeferred>() + mockSyncWorkflowDraft + .mockReturnValueOnce(pending.promise) + .mockReturnValueOnce(unloading.promise) + .mockImplementationOnce(async ({ params }) => ({ + hash: 'recovered-save', + updated_at: 3, + client_save_id: params.client_save_id, + })) + const variables: Shape['environmentVariables'] = [ + { id: 'env-1', name: 'topic', value_type: 'string', value: 'Queued edit', description: '' }, + ] + const { result, store, setCanvas } = renderUseNodesSyncDraft() + const saving = result.current.doSyncWorkflowDraft() + await waitFor(() => expect(mockSyncWorkflowDraft).toHaveBeenCalledOnce()) + store.getState().setEnvironmentVariables(variables) + const queued = result.current.doSyncWorkflowDraft(false, undefined, { + environmentVariablePatch: { + environmentVariables: variables, + deletedEnvironmentVariableIds: ['removed'], + }, + }) + setCanvas({ nodes: [createNode({ id: 'unload-edit' })] }) + result.current.syncWorkflowDraftWhenPageClose() + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + const unloadRequest = mockSyncWorkflowDraft.mock.calls[1]?.[0].params + expect(unloadRequest?.predecessor_saves).toEqual([ + { client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id }, + { + client_save_id: expect.any(String), + environment_variable_patch: { + environment_variables: variables, + deleted_environment_variable_ids: ['removed'], + }, + }, + ]) + await act(async () => { + unloading.reject(new TypeError('Failed to fetch')) + pending.resolve({ + hash: 'older-save', + updated_at: 1, + client_save_id: mockSyncWorkflowDraft.mock.calls[0]?.[0].params.client_save_id, + }) + await Promise.all([saving, queued]) + await store.getState().enqueueWorkflowDraftOperation(async () => undefined) + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(2) + expect(store.getState().hasWorkflowDraftConflict).toBe(false) + setCanvas({ nodes: [createNode({ id: 'latest-retry-edit' })] }) + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(3) + const retryRequest = mockSyncWorkflowDraft.mock.calls[2]?.[0].params + expect(retryRequest).toMatchObject({ + predecessor_saves: [ + ...(unloadRequest?.predecessor_saves ?? []), + { client_save_id: unloadRequest?.client_save_id }, + ], + graph: { + nodes: [expect.objectContaining({ id: 'latest-retry-edit' })], + }, + }) + expect(retryRequest?.environment_variable_patch).toBeUndefined() + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'recovered-save', + syncWorkflowDraftSaveId: retryRequest?.client_save_id, + environmentVariables: variables, + hasWorkflowDraftConflict: false, + }) + await act(async () => { + await result.current.doSyncWorkflowDraft() + }) + expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(4) + const nextRequest = mockSyncWorkflowDraft.mock.calls[3]?.[0].params + expect(nextRequest?.base_client_save_id).toBe(retryRequest?.client_save_id) + expect(nextRequest?.predecessor_saves).toBeUndefined() }) it.each(['autosave', 'builder'] as const)( diff --git a/web/app/components/workflow-app/hooks/__tests__/use-workflow-init.spec.ts b/web/app/components/workflow-app/hooks/__tests__/use-workflow-init.spec.ts index 53575c14c60..3c28112abdd 100644 --- a/web/app/components/workflow-app/hooks/__tests__/use-workflow-init.spec.ts +++ b/web/app/components/workflow-app/hooks/__tests__/use-workflow-init.spec.ts @@ -113,6 +113,7 @@ const draftResponse = { edges: [], }, hash: 'server-hash', + client_save_id: '00000000-0000-4000-8000-000000000001', created_at: 0, created_by: { id: '', name: '', email: '' }, updated_at: 1, @@ -193,6 +194,7 @@ describe('useWorkflowInit', () => { }), ) expect(store.getState().syncWorkflowDraftHash).toBe('new-workflow-hash') + expect(store.getState().syncWorkflowDraftSaveId).toBe(draftResponse.client_save_id) expect(store.getState().syncWorkflowDraftIsEmpty).toBe(true) }) @@ -242,6 +244,7 @@ describe('useWorkflowInit', () => { const { result, store } = renderHook(() => useWorkflowInit(), { syncWorkflowDraftHash: 'stale-hash', + syncWorkflowDraftSaveId: '00000000-0000-4000-8000-000000000002', }) await waitFor(() => { @@ -258,6 +261,7 @@ describe('useWorkflowInit', () => { conversationVariables: [], isWorkflowDataLoaded: true, syncWorkflowDraftHash: '', + syncWorkflowDraftSaveId: null, syncWorkflowDraftIsEmpty: true, }) expect(result.current.data?.graph.viewport).toBeUndefined() @@ -352,6 +356,7 @@ describe('useWorkflowInit', () => { isWorkflowDataLoaded: true, nodesDefaultConfigs: { start: { title: 'Start Config' } }, syncWorkflowDraftHash: 'server-hash', + syncWorkflowDraftSaveId: draftResponse.client_save_id, draftUpdatedAt: 9000, toolPublished: true, lastPublishedHasUserInput: true, diff --git a/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.builder.spec.ts b/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.builder.spec.ts index c7fe2e0d869..10d1fef8c94 100644 --- a/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.builder.spec.ts +++ b/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.builder.spec.ts @@ -1,6 +1,7 @@ import type { AppMode } from '@dify/contracts/api/console/apps/types.gen' import type { ReactNode } from 'react' import type { WorkflowDraftRefreshResult } from '../use-workflow-refresh-draft' +import type { ServerDraftUpdate } from '@/app/components/workflow/collaboration/core/server-draft-sync' import type { ConversationVariable } from '@/app/components/workflow/types' import { act, screen, waitFor } from '@testing-library/react' import userEvent from '@testing-library/user-event' @@ -38,7 +39,7 @@ vi.mock('@/app/components/workflow/hooks/use-workflow-update', () => ({ vi.mock('@/service/workflow', () => ({ fetchWorkflowDraft: fetchDraft })) const collaboration = vi.hoisted(() => ({ connected: vi.fn(() => false), - sync: vi.fn(), + sync: vi.fn<() => Promise>>(), nodes: vi.fn(), edges: vi.fn(() => []), })) @@ -54,6 +55,7 @@ vi.mock('@/app/components/workflow/collaboration/core/collaboration-manager', () const draftResponse = { hash: 'server-hash', + client_save_id: '00000000-0000-4000-8000-000000000001', graph: { nodes: [createNode({ id: 'n1', data: { type: BlockEnum.Start } })], edges: [], @@ -275,7 +277,10 @@ describe('useWorkflowRefreshDraft', () => { collaboration.connected.mockReturnValue(true) const accepted = [createNode({ id: 'accepted' })] collaboration.nodes.mockReturnValue(accepted) - collaboration.sync.mockResolvedValue({ hash: 'server-hash' }) + collaboration.sync.mockResolvedValue({ + hash: 'server-hash', + client_save_id: draftResponse.client_save_id, + }) const { result, store } = setup() await act(async () => expect(await result.current.handleRefreshWorkflowDraft()).toBe('refreshed'), @@ -389,8 +394,16 @@ describe('useWorkflowRefreshDraft', () => { }) it('reads matching metadata when collaboration has accepted a newer draft', async () => { collaboration.connected.mockReturnValue(true) - collaboration.sync.mockResolvedValue({ hash: 'newer-hash' }) - const latest = { ...draftResponse, hash: 'newer-hash', conversation_variables: [] } + const latest = { + ...draftResponse, + hash: 'newer-hash', + client_save_id: '00000000-0000-4000-8000-000000000002', + conversation_variables: [], + } + collaboration.sync.mockResolvedValue({ + hash: latest.hash, + client_save_id: latest.client_save_id, + }) fetchDraft.mockResolvedValueOnce(draftResponse).mockResolvedValueOnce(latest) const { result, store } = setup() await act(async () => @@ -399,11 +412,45 @@ describe('useWorkflowRefreshDraft', () => { expect(fetchDraft).toHaveBeenCalledTimes(2) expect(updateCanvas).toHaveBeenCalledOnce() expect(store.getState().syncWorkflowDraftHash).toBe('newer-hash') + expect(store.getState().syncWorkflowDraftSaveId).toBe(latest.client_save_id) + }) + + it('refetches metadata when collaboration advances only the save receipt', async () => { + const latest = { + ...draftResponse, + client_save_id: '00000000-0000-4000-8000-000000000002', + environment_variables: [ + { id: 'env-1', name: 'topic', value_type: 'string', value: 'Remote edit', description: '' }, + ], + } + collaboration.connected.mockReturnValue(true) + collaboration.sync.mockResolvedValue({ + hash: latest.hash, + client_save_id: latest.client_save_id, + }) + collaboration.nodes.mockReturnValue(latest.graph.nodes) + fetchDraft.mockResolvedValueOnce(draftResponse).mockResolvedValueOnce(latest) + const { result, store } = setup() + + await act(async () => + expect(await result.current.handleRefreshWorkflowDraft()).toBe('refreshed'), + ) + + expect(fetchDraft).toHaveBeenCalledTimes(2) + expect(updateCanvas).toHaveBeenCalledOnce() + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: latest.hash, + syncWorkflowDraftSaveId: latest.client_save_id, + environmentVariables: latest.environment_variables, + }) }) it('finishes the Builder refresh without manual Retry when collaboration advances after the HTTP read', async () => { collaboration.connected.mockReturnValue(true) - collaboration.sync.mockResolvedValue({ hash: 'newer-hash' }) + collaboration.sync.mockResolvedValue({ + hash: 'newer-hash', + client_save_id: draftResponse.client_save_id, + }) collaboration.nodes.mockReturnValue(draftResponse.graph.nodes) fetchDraft .mockResolvedValueOnce(draftResponse) diff --git a/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.spec.tsx b/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.spec.tsx index 39340150370..c071d2b69b5 100644 --- a/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.spec.tsx +++ b/web/app/components/workflow-app/hooks/__tests__/use-workflow-refresh-draft.spec.tsx @@ -82,9 +82,11 @@ const savedNodes = [ ] const savedEdges = [createEdge({ id: 'saved-edge', source: 'saved-start', target: 'saved-code' })] const savedViewport = { x: 10, y: 20, zoom: 1.2 } +const localClientSaveId = '00000000-0000-4000-8000-000000000001' const draftResponse: FetchWorkflowDraftResponse = { id: 'draft-1', hash: 'server-hash', + client_save_id: '00000000-0000-4000-8000-000000000002', graph: { nodes: savedNodes, edges: savedEdges, viewport: savedViewport }, created_at: 1, created_by: { id: 'user-1', name: 'User', email: 'user@example.com' }, @@ -144,6 +146,7 @@ function setup(mode: AppMode = 'advanced-chat') { nodes: localNodes, isWorkflowDataLoaded: true, syncWorkflowDraftHash: 'local-hash', + syncWorkflowDraftSaveId: localClientSaveId, syncWorkflowDraftIsEmpty: false, }, }, @@ -201,6 +204,7 @@ describe('useWorkflowRefreshDraft', () => { ) expect(store.getState()).toMatchObject({ syncWorkflowDraftHash: notUpdateCanvas ? 'local-hash' : 'server-hash', + syncWorkflowDraftSaveId: notUpdateCanvas ? localClientSaveId : draftResponse.client_save_id, syncWorkflowDraftIsEmpty: !!notUpdateCanvas, isSyncingWorkflowDraft: false, isWorkflowDataLoaded: true, @@ -229,6 +233,7 @@ describe('useWorkflowRefreshDraft', () => { await expectGraph(flow, ['local-start', 'local-code'], ['local-edge'], localViewport) expect(store.getState()).toMatchObject({ syncWorkflowDraftHash: 'local-hash', + syncWorkflowDraftSaveId: localClientSaveId, syncWorkflowDraftIsEmpty: false, environmentVariables: [], conversationVariables: [], @@ -306,6 +311,7 @@ describe('useWorkflowRefreshDraft', () => { ] satisfies NonNullable fetchDraft.mockResolvedValue({ ...draftResponse, + client_save_id: null, graph: { nodes: savedNodes, edges: savedEdges }, environment_variables: [ { @@ -339,6 +345,7 @@ describe('useWorkflowRefreshDraft', () => { zoom: 1, }) expect(store.getState()).toMatchObject({ + syncWorkflowDraftSaveId: null, envSecrets: { 'env-secret': 'top-secret' }, environmentVariables: [ { @@ -413,6 +420,7 @@ describe('useWorkflowRefreshDraft', () => { isWorkflowDataLoaded: true, isSyncingWorkflowDraft: false, syncWorkflowDraftHash: 'local-hash', + syncWorkflowDraftSaveId: localClientSaveId, syncWorkflowDraftIsEmpty: false, }) await expectGraph(flow, ['local-start', 'local-code'], ['local-edge'], localViewport) diff --git a/web/app/components/workflow-app/hooks/use-nodes-sync-draft.ts b/web/app/components/workflow-app/hooks/use-nodes-sync-draft.ts index ebba5ff36e4..caa0e16bcec 100644 --- a/web/app/components/workflow-app/hooks/use-nodes-sync-draft.ts +++ b/web/app/components/workflow-app/hooks/use-nodes-sync-draft.ts @@ -4,7 +4,9 @@ import type { SyncDraftOptions, SyncDraftResult, } from '@/app/components/workflow/hooks-store' -import { safe } from '@orpc/client' +import type { WorkflowDraftSave } from '@/app/components/workflow/store/workflow/workflow-draft-slice' +import type { EnvironmentVariablePatch } from '@/service/workflow' +import { ORPCError, safe } from '@orpc/client' import { skipToken, useQuery, useSuspenseQuery } from '@tanstack/react-query' import { produce } from 'immer' import { useStore as useAtomStore, useAtomValue } from 'jotai' @@ -44,6 +46,17 @@ const shouldSkipDraftSync = ( const isEmptyGraph = (graph: { nodes: unknown[]; edges: unknown[] }) => graph.nodes.length === 0 && graph.edges.length === 0 +const toEnvironmentVariablePatchPayload = (patch: EnvironmentVariablePatch) => ({ + environment_variables: patch.environmentVariables, + deleted_environment_variable_ids: patch.deletedEnvironmentVariableIds, +}) + +type LocalDraftSyncOptions = { + preparingBuilder?: boolean + force?: boolean + keepalive?: boolean +} + const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { const { t } = useTranslation(['workflow']) const store = useStoreApi() @@ -98,6 +111,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { const { conversationVariables, syncWorkflowDraftHash, + syncWorkflowDraftSaveId, isWorkflowDataLoaded, isSyncingWorkflowDraft, hasWorkflowDraftConflict, @@ -156,6 +170,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { appId, generation: workflowDraftGeneration, localRevision: workflowDraftLocalRevision, + baseClientSaveId: syncWorkflowDraftSaveId, url: `/apps/${appId}/workflows/draft`, params: { graph: { @@ -178,13 +193,9 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { const performLocalSync = useCallback( async ( baseParams: NonNullable>, + save: WorkflowDraftSave, callback?: SyncDraftCallback, - options?: SyncDraftOptions, - { - preparingBuilder = false, - force = false, - keepalive = false, - }: { preparingBuilder?: boolean; force?: boolean; keepalive?: boolean } = {}, + { preparingBuilder = false, force = false, keepalive = false }: LocalDraftSyncOptions = {}, ): Promise => { if ( !preparingBuilder && @@ -196,6 +207,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { const isCurrent = () => { const state = workflowStore.getState() return ( + save.isCurrent() && atomStore.get(workflowAppIdentityAtom).appId === baseParams.appId && state.workflowDraftGeneration === baseParams.generation && !shouldSkipDraftSync( @@ -223,6 +235,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { } let expectedHash = baseParams.params.hash + let expectedSaveId = baseParams.baseClientSaveId const emptyGraph = isEmptyGraph(baseParams.params.graph) const canSaveEmptyGraph = () => { const currentParams = getPostParams() @@ -242,7 +255,10 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { if (emptyGraph && force && isCollaborationEnabled) { // CRDT updates do not carry the persisted revision from another collaborator's save. - const hashBeforeRefresh = workflowStore.getState().syncWorkflowDraftHash + const { + syncWorkflowDraftHash: hashBeforeRefresh, + syncWorkflowDraftSaveId: saveIdBeforeRefresh, + } = workflowStore.getState() const [error, draft] = await safe( consoleClient.apps.byAppId.workflows.draft.get( { params: { app_id: baseParams.appId } }, @@ -259,14 +275,20 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { return null } - const { syncWorkflowDraftHash, setSyncWorkflowDraftHash } = workflowStore.getState() + const { syncWorkflowDraftHash, syncWorkflowDraftSaveId, setSyncWorkflowDraftHash } = + workflowStore.getState() // Another hook may have saved a newer revision while this request was in flight. - if (syncWorkflowDraftHash === hashBeforeRefresh) { + if ( + syncWorkflowDraftHash === hashBeforeRefresh && + syncWorkflowDraftSaveId === saveIdBeforeRefresh + ) { expectedHash = draft.hash - setSyncWorkflowDraftHash(draft.hash, { - nodes: draft.graph.nodes, - edges: draft.graph.edges, - }) + expectedSaveId = draft.client_save_id ?? null + setSyncWorkflowDraftHash( + draft.hash, + { nodes: draft.graph.nodes, edges: draft.graph.edges }, + draft.client_save_id, + ) } } @@ -305,28 +327,48 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { const { setSyncWorkflowDraftHash, setDraftUpdatedAt } = workflowStore.getState() try { - const currentHash = workflowStore.getState().syncWorkflowDraftHash + const { syncWorkflowDraftHash: currentHash, syncWorkflowDraftSaveId: currentSaveId } = + workflowStore.getState() const lastSave = workflowStore.getState().lastSavedWorkflowDraft - // Only a successful save from this queue may advance a captured graph's + // Only a successful save from this editor may advance a captured graph's // hash. A server refresh must never lend its hash to an older graph. const canUseLastSave = lastSave !== null && lastSave.appId === baseParams.appId && lastSave.generation === baseParams.generation && - lastSave.hash === currentHash + lastSave.hash === currentHash && + lastSave.clientSaveId === currentSaveId if (currentHash !== expectedHash && !canUseLastSave) return null + const submission = save.start() + if (!submission) return null + + const baseClientSaveId = canUseLastSave ? lastSave.clientSaveId : expectedSaveId const body = { ...baseParams.params, hash: (canUseLastSave ? lastSave.hash : expectedHash) || null, - ...(force ? { force: true } : {}), - ...(options?.environmentVariablePatch + client_save_id: submission.clientSaveId, + ...(baseClientSaveId ? { base_client_save_id: baseClientSaveId } : {}), + ...(submission.predecessorSaves.length > 0 ? { - environment_variable_patch: { - environment_variables: options.environmentVariablePatch.environmentVariables, - deleted_environment_variable_ids: - options.environmentVariablePatch.deletedEnvironmentVariableIds, - }, + predecessor_saves: submission.predecessorSaves.map((predecessor) => ({ + client_save_id: predecessor.clientSaveId, + ...(predecessor.environmentVariablePatch + ? { + environment_variable_patch: toEnvironmentVariablePatchPayload( + predecessor.environmentVariablePatch, + ), + } + : {}), + })), + } + : {}), + ...(force ? { force: true } : {}), + ...(submission.environmentVariablePatch + ? { + environment_variable_patch: toEnvironmentVariablePatchPayload( + submission.environmentVariablePatch, + ), } : {}), } @@ -339,22 +381,36 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { ) if (!isCurrent()) return null if (error) { + if ( + error instanceof ORPCError && + error.status >= 400 && + error.status < 500 && + error.status !== 408 && + error.status !== 429 + ) + save.reject() if (isDefined && error.code === 'draft_workflow_not_sync') workflowStore.getState().setWorkflowDraftConflict(true) callback?.onError?.() return null } - if (workflowStore.getState().syncWorkflowDraftHash !== currentHash) return null + if ( + workflowStore.getState().syncWorkflowDraftHash !== currentHash || + workflowStore.getState().syncWorkflowDraftSaveId !== currentSaveId + ) + return null + save.acknowledge() workflowStore.getState().setLastSavedWorkflowDraft({ appId: baseParams.appId, generation: baseParams.generation, hash: res.hash, + clientSaveId: res.client_save_id ?? null, }) - setSyncWorkflowDraftHash(res.hash, baseParams.params.graph) + setSyncWorkflowDraftHash(res.hash, baseParams.params.graph, res.client_save_id) setDraftUpdatedAt(res.updated_at) workflowStore.getState().markWorkflowDraftSaved(baseParams.localRevision) callback?.onSuccess?.() - return { hash: res.hash, updatedAt: res.updated_at } + return { hash: res.hash, updatedAt: res.updated_at, clientSaveId: res.client_save_id } } finally { callback?.onSettled?.() } @@ -363,16 +419,27 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { ) const doSyncWorkflowDraftLocally = useCallback( - (...args: Parameters) => - workflowStore.getState().enqueueWorkflowDraftOperation(() => { - // Saves accepted before the barrier must finish, including variable - // patches that cannot be reconstructed from the final canvas snapshot. - const drainingForBuilder = workflowStore.getState().workflowDraftSyncPhase !== 'idle' - return performLocalSync(args[0], args[1], args[2], { - ...args[3], - preparingBuilder: args[3]?.preparingBuilder || drainingForBuilder, - }) - }), + ( + baseParams: NonNullable>, + callback?: SyncDraftCallback, + options?: SyncDraftOptions, + syncOptions: LocalDraftSyncOptions = {}, + ) => + workflowStore.getState().runWorkflowDraftSave( + { + appId: baseParams.appId, + generation: baseParams.generation, + keepalive: syncOptions.keepalive === true, + environmentVariablePatch: options?.environmentVariablePatch, + }, + (save) => { + const drainingForBuilder = workflowStore.getState().workflowDraftSyncPhase !== 'idle' + return performLocalSync(baseParams, save, callback, { + ...syncOptions, + preparingBuilder: syncOptions.preparingBuilder || drainingForBuilder, + }) + }, + ), [performLocalSync, workflowStore], ) @@ -386,10 +453,11 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { if (!canPersist) return const snapshot = getPostParams() - // Keep unload requests in the same queue so a returning tab and Builder - // observe their acknowledged hash. Empty-canvas saves still need consent. + // Dispatch before teardown; queued reads and Builder still await this save. if (snapshot && !isEmptyGraph(snapshot.params.graph)) - void doSyncWorkflowDraftLocally(snapshot, undefined, undefined, { keepalive: true }) + void doSyncWorkflowDraftLocally(snapshot, undefined, undefined, { keepalive: true }).catch( + () => undefined, + ) }, [ doSyncWorkflowDraftLocally, getNodesReadOnly, @@ -473,7 +541,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { ) return null const { setSyncWorkflowDraftHash, setDraftUpdatedAt } = workflowStore.getState() - setSyncWorkflowDraftHash(result.hash, baseParams.params.graph) + setSyncWorkflowDraftHash(result.hash, baseParams.params.graph, result.clientSaveId) setDraftUpdatedAt(result.updatedAt) workflowStore.getState().markWorkflowDraftSaved(workflowDraftLocalRevision) callback?.onSuccess?.() @@ -535,11 +603,11 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { try { let saved: SyncDraftResult | null = null if (shouldSave && !requestLeader) { - saved = await state.enqueueWorkflowDraftOperation(async () => - snapshot - ? performLocalSync(snapshot, undefined, undefined, { preparingBuilder: true }) - : null, - ) + saved = snapshot + ? await doSyncWorkflowDraftLocally(snapshot, undefined, undefined, { + preparingBuilder: true, + }) + : await state.enqueueWorkflowDraftOperation(async () => null) } else { await state.enqueueWorkflowDraftOperation(async () => undefined) signal.throwIfAborted() @@ -554,7 +622,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { workflowStore.getState().workflowDraftGeneration !== generation ) throw new Error(t(($) => $['common.draftSaveFailed'])) - state.setSyncWorkflowDraftHash(saved.hash) + state.setSyncWorkflowDraftHash(saved.hash, undefined, saved.clientSaveId) state.setDraftUpdatedAt(saved.updatedAt) } } @@ -580,10 +648,10 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => { [ appId, atomStore, + doSyncWorkflowDraftLocally, getNodesReadOnly, getPostParams, isCollaborationEnabled, - performLocalSync, t, workflowStore, ], diff --git a/web/app/components/workflow-app/hooks/use-workflow-init.ts b/web/app/components/workflow-app/hooks/use-workflow-init.ts index d1f880741b4..f666ecaa0de 100644 --- a/web/app/components/workflow-app/hooks/use-workflow-init.ts +++ b/web/app/components/workflow-app/hooks/use-workflow-init.ts @@ -121,7 +121,7 @@ export const useWorkflowInit = () => { conversationVariables: initialData.conversation_variables || [], isWorkflowDataLoaded: true, }) - setSyncWorkflowDraftHash(initialData.hash, res.graph) + setSyncWorkflowDraftHash(initialData.hash, res.graph, res.client_save_id) setIsLoading(false) } catch (error: unknown) { if (error instanceof ORPCError) { @@ -175,7 +175,7 @@ export const useWorkflowInit = () => { ) .then((res) => { workflowStore.getState().setDraftUpdatedAt(res.updated_at) - setSyncWorkflowDraftHash(res.hash, initialGraph) + setSyncWorkflowDraftHash(res.hash, initialGraph, res.client_save_id) handleGetInitialWorkflowData() }) } diff --git a/web/app/components/workflow-app/hooks/use-workflow-refresh-draft.ts b/web/app/components/workflow-app/hooks/use-workflow-refresh-draft.ts index 544f4bbd3bb..ce47e4875cd 100644 --- a/web/app/components/workflow-app/hooks/use-workflow-refresh-draft.ts +++ b/web/app/components/workflow-app/hooks/use-workflow-refresh-draft.ts @@ -130,7 +130,11 @@ export const useWorkflowRefreshDraft = () => { if (!canApply()) return 'superseded' // The socket may already have accepted a newer committed draft. // Read matching metadata before applying either graph or hash. - if (synced.hash !== response.hash) continue + if ( + synced.hash !== response.hash || + (synced.client_save_id ?? null) !== (response.client_save_id ?? null) + ) + continue graph = { ...graph, nodes: collaborationManager.getNodes(), @@ -145,7 +149,7 @@ export const useWorkflowRefreshDraft = () => { }) if (!applied) return 'failed' // The hash is a baseline for this graph, never for a metadata-only refresh. - setSyncWorkflowDraftHash(response.hash, response.graph) + setSyncWorkflowDraftHash(response.hash, response.graph, response.client_save_id) } setEnvSecrets( (response.environment_variables || []) diff --git a/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx b/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx index 7b8c817181f..834c4f152ff 100644 --- a/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx +++ b/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx @@ -169,6 +169,7 @@ describe('UpdateDSLModal', () => { graph: { nodes: [importedNode], edges: [], viewport: { x: 0, y: 0, zoom: 1 } }, features: {}, hash: 'imported-hash', + client_save_id: '40c6cf5c-a5a1-4bfd-b3ae-f072e3f1efcd', conversation_variables: [], environment_variables: [], }) @@ -190,7 +191,12 @@ describe('UpdateDSLModal', () => { expect(mockEmit).toHaveBeenCalledWith( expect.objectContaining({ type: 'WORKFLOW_DATA_UPDATE', - payload: expect.objectContaining({ nodes, edges, hash: 'imported-hash' }), + payload: expect.objectContaining({ + nodes, + edges, + hash: 'imported-hash', + client_save_id: '40c6cf5c-a5a1-4bfd-b3ae-f072e3f1efcd', + }), }), ) expect(mockReplaceGraphFromCommittedDraft.mock.invocationCallOrder[0]).toBeLessThan( diff --git a/web/app/components/workflow/__tests__/workflow-edge-events.spec.tsx b/web/app/components/workflow/__tests__/workflow-edge-events.spec.tsx index eedb1d1240b..5948c210f2a 100644 --- a/web/app/components/workflow/__tests__/workflow-edge-events.spec.tsx +++ b/web/app/components/workflow/__tests__/workflow-edge-events.spec.tsx @@ -192,12 +192,18 @@ it('opens the node context menu and clears its target when the graph is replaced }), ], edges: [], + hash: 'imported-hash', + client_save_id: '40c6cf5c-a5a1-4bfd-b3ae-f072e3f1efcd', }, }), ) expect(store.getState().contextMenuTarget).toBeUndefined() expect(screen.getByText('Imported graph')).toBeInTheDocument() expect(screen.queryByText('Transform documents')).not.toBeInTheDocument() + expect(store.getState()).toMatchObject({ + syncWorkflowDraftHash: 'imported-hash', + syncWorkflowDraftSaveId: '40c6cf5c-a5a1-4bfd-b3ae-f072e3f1efcd', + }) }) it('reports a persistent error when the final save fails', async () => { diff --git a/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.socket-and-subscriptions.spec.ts b/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.socket-and-subscriptions.spec.ts index 0748054afae..9a60838b456 100644 --- a/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.socket-and-subscriptions.spec.ts +++ b/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.socket-and-subscriptions.spec.ts @@ -85,21 +85,24 @@ it('sends cursor and workflow changes only while connected', async () => { expect(emit).not.toHaveBeenCalled() }) -it('includes a graph snapshot and resolves the workflow save acknowledgement', async () => { - const { manager, messages, acknowledge } = await createSession({ nodes: [createNode('saved')] }) - const saved = manager.requestWorkflowSync() - expect(messages('sync_request')).toEqual([ - [ - 'collaboration_event', - expect.objectContaining({ - data: { requestId: expect.any(String), graphSnapshot: expect.any(Uint8Array) }, - }), - expect.any(Function), - ], - ]) - acknowledge('sync_request', { success: true, hash: 'hash-2', updatedAt: 2 }, 200) - await expect(saved).resolves.toEqual({ hash: 'hash-2', updatedAt: 2 }) -}) +it.each([undefined, null, '019b0766-9540-4dd9-a51f-bf7650f4b7bd'])( + 'includes a graph snapshot and resolves the workflow save acknowledgement with receipt %s', + async (clientSaveId) => { + const { manager, messages, acknowledge } = await createSession({ nodes: [createNode('saved')] }) + const saved = manager.requestWorkflowSync() + expect(messages('sync_request')).toEqual([ + [ + 'collaboration_event', + expect.objectContaining({ + data: { requestId: expect.any(String), graphSnapshot: expect.any(Uint8Array) }, + }), + expect.any(Function), + ], + ]) + acknowledge('sync_request', { success: true, hash: 'hash-2', updatedAt: 2, clientSaveId }, 200) + await expect(saved).resolves.toEqual({ hash: 'hash-2', updatedAt: 2, clientSaveId }) + }, +) it('rejects a workflow save when the server reports failure', async () => { const { manager, acknowledge } = await createSession() @@ -126,12 +129,18 @@ it('imports a directed save snapshot even when the local leader flag is stale an expect(requests).toHaveBeenCalledOnce() const request = requests.mock.calls[0]![0] expect(request.requestId).toBe('request-1') - request.acknowledge({ success: true, hash: 'hash-1', updatedAt: 1 }) + request.acknowledge({ + success: true, + hash: 'hash-1', + updatedAt: 1, + clientSaveId: '019b0766-9540-4dd9-a51f-bf7650f4b7bd', + }) request.acknowledge({ success: false, error: 'duplicate' }) expect(acknowledge).toHaveBeenCalledExactlyOnceWith({ success: true, hash: 'hash-1', updatedAt: 1, + clientSaveId: '019b0766-9540-4dd9-a51f-bf7650f4b7bd', }) }) @@ -383,6 +392,39 @@ it('keeps a promoted follower blocked until an authoritative snapshot or reload expect(manager.canPersistLocalGraph()).toBe(false) }) +it('delivers the promoted saver baseline before notifying listeners that saving is ready', async () => { + const remote = await createSession({ appId: 'remote', nodes: [createNode('shared')] }) + const { manager, receive } = await createSession({ role: 'follower' }) + receive('graph_update', remote.snapshot()) + expect(manager.canPersistLocalGraph()).toBe(true) + const baseline = vi.fn() + const leader = vi.fn() + const ready = vi.fn() + const reload = vi.fn() + const unsubscribe = manager.onDraftBaseline(baseline) + manager.onLeaderChange(leader) + manager.onGraphReadyChange(ready) + manager.onGraphReloadRequired(reload) + ready.mockClear() + const draftBaseline = { + hash: 'shared-hash', + client_save_id: '00000000-0000-4000-8000-000000000002', + } + + receive('status', { isLeader: true, draftBaseline }) + + expect(baseline).toHaveBeenCalledExactlyOnceWith(draftBaseline) + expect(leader).toHaveBeenCalledExactlyOnceWith(true) + expect(ready).toHaveBeenCalledWith(true) + expect(baseline.mock.invocationCallOrder[0]).toBeLessThan(leader.mock.invocationCallOrder[0]!) + expect(baseline.mock.invocationCallOrder[0]).toBeLessThan(ready.mock.invocationCallOrder[0]!) + expect(reload).not.toHaveBeenCalled() + expect(manager.canFlushGraphOnPageClose()).toBe(true) + unsubscribe() + receive('status', { isLeader: true, draftBaseline: { ...draftBaseline, client_save_id: null } }) + expect(baseline).toHaveBeenCalledOnce() +}) + it('retries follower resync until a snapshot arrives and then stops', async () => { vi.useFakeTimers() const remote = await createSession({ appId: 'remote', nodes: [createNode('snapshot')] }) diff --git a/web/app/components/workflow/collaboration/core/collaboration-manager.ts b/web/app/components/workflow/collaboration/core/collaboration-manager.ts index d9c7f8cbfc8..1c340c2addc 100644 --- a/web/app/components/workflow/collaboration/core/collaboration-manager.ts +++ b/web/app/components/workflow/collaboration/core/collaboration-manager.ts @@ -13,6 +13,7 @@ import type { OnlineUser, RestoreCompleteData, RestoreIntentData, + WorkflowDraftBaseline, WorkflowSyncAcknowledgement, WorkflowSyncRequest, WorkflowSyncResult, @@ -1098,6 +1099,7 @@ export class CollaborationManager { success?: unknown hash?: unknown updatedAt?: unknown + clientSaveId?: unknown error?: unknown msg?: unknown } @@ -1116,7 +1118,13 @@ export class CollaborationManager { return } - pendingRequest.resolve({ hash: response.hash, updatedAt: response.updatedAt }) + pendingRequest.resolve({ + hash: response.hash, + updatedAt: response.updatedAt, + ...(typeof response.clientSaveId === 'string' || response.clientSaveId === null + ? { clientSaveId: response.clientSaveId } + : {}), + }) }, onUnauthorized: this.handleSessionUnauthorized, }, @@ -1240,6 +1248,10 @@ export class CollaborationManager { return this.eventEmitter.on('leaderChange', callback) } + onDraftBaseline(callback: (baseline: WorkflowDraftBaseline) => void): () => void { + return this.eventEmitter.on('draftBaseline', callback) + } + onCommentsUpdate(callback: (update: { appId: string; timestamp: number }) => void): () => void { return this.eventEmitter.on('commentsUpdate', callback) } @@ -1999,7 +2011,7 @@ export class CollaborationManager { } }) - socket.on('status', (data: { isLeader: boolean }) => { + socket.on('status', (data: { isLeader: boolean; draftBaseline?: WorkflowDraftBaseline }) => { try { if (!data || typeof data.isLeader !== 'boolean') { console.warn('Invalid status data:', data) @@ -2009,6 +2021,8 @@ export class CollaborationManager { const wasLeader = this.isLeader const wasAwaitingSnapshotImport = this.awaitingSnapshotImport this.isLeader = data.isLeader + if (this.isLeader && data.draftBaseline) + this.eventEmitter.emit('draftBaseline', data.draftBaseline) if ( this.isLeader && diff --git a/web/app/components/workflow/collaboration/core/server-draft-sync.ts b/web/app/components/workflow/collaboration/core/server-draft-sync.ts index 75c3037e0a3..e4b08d15792 100644 --- a/web/app/components/workflow/collaboration/core/server-draft-sync.ts +++ b/web/app/components/workflow/collaboration/core/server-draft-sync.ts @@ -1,9 +1,10 @@ +import type { WorkflowResponse } from '@dify/contracts/api/console/apps/types.gen' import type { Socket } from 'socket.io-client' import type { Edge, Node } from '../../types' import { emitWithAuthGuard } from './websocket-manager' export type ServerDraftGraph = { nodes: Node[]; edges: Edge[] } -export type ServerDraftChange = { +export type ServerDraftChange = Pick & { revision: string hash: string updated_at: number @@ -11,7 +12,10 @@ export type ServerDraftChange = { previous_graph: ServerDraftGraph | null base_update: string | null } -export type ServerDraftUpdate = Pick & { +export type ServerDraftUpdate = Pick< + ServerDraftChange, + 'revision' | 'hash' | 'updated_at' | 'client_save_id' +> & { update: Uint8Array } diff --git a/web/app/components/workflow/collaboration/types/collaboration.ts b/web/app/components/workflow/collaboration/types/collaboration.ts index c8f788b4c15..8dbfc1a8713 100644 --- a/web/app/components/workflow/collaboration/types/collaboration.ts +++ b/web/app/components/workflow/collaboration/types/collaboration.ts @@ -1,3 +1,8 @@ +import type { + SyncDraftWorkflowResponse, + WorkflowResponse, +} from '@dify/contracts/api/console/apps/types.gen' + export type OnlineUser = { user_id: string username: string @@ -60,8 +65,11 @@ export type CollaborationUpdate = { export type WorkflowSyncResult = { hash: string updatedAt: number + clientSaveId?: SyncDraftWorkflowResponse['client_save_id'] } +export type WorkflowDraftBaseline = Pick + export type WorkflowSyncAcknowledgement = | ({ success: true } & WorkflowSyncResult) | { success: false; error?: string } diff --git a/web/app/components/workflow/hooks-store/store.ts b/web/app/components/workflow/hooks-store/store.ts index 2d5a7e96a32..2db346ffbbf 100644 --- a/web/app/components/workflow/hooks-store/store.ts +++ b/web/app/components/workflow/hooks-store/store.ts @@ -1,3 +1,4 @@ +import type { SyncDraftWorkflowResponse } from '@dify/contracts/api/console/apps/types.gen' import type { SchemaDefinitionItemResponse } from '@dify/contracts/api/console/spec/types.gen' import type { FileUpload } from '../../base/features/types' import type { @@ -29,6 +30,7 @@ export type SyncDraftCallback = { export type SyncDraftResult = { hash: string updatedAt: number + clientSaveId?: SyncDraftWorkflowResponse['client_save_id'] } export type SyncDraftOptions = { diff --git a/web/app/components/workflow/index.tsx b/web/app/components/workflow/index.tsx index c5f5301ba07..e2a2c9f2a6f 100644 --- a/web/app/components/workflow/index.tsx +++ b/web/app/components/workflow/index.tsx @@ -8,7 +8,7 @@ import type { WorkflowHistoryState } from './store/workflow/history-slice' import type { WorkflowSliceShape } from './store/workflow/workflow-slice' import type { ConversationVariable, Edge, EnvironmentVariable, Node } from './types' import type { EventEmitterValue } from '@/context/event-emitter' -import type { VarInInspect } from '@/types/workflow' +import type { FetchWorkflowDraftResponse, VarInInspect } from '@/types/workflow' import { AlertDialog, AlertDialogCancelButton, @@ -126,7 +126,7 @@ const WorkflowNodeSearch = memo(() => { return null }) -type WorkflowDataUpdatePayload = { +type WorkflowDataUpdatePayload = Pick & { nodes: Node[] edges: Edge[] viewport?: Viewport @@ -319,7 +319,7 @@ export const Workflow: FC = memo( if (payload.viewport) reactflow.setViewport(payload.viewport) - if (payload.hash) setSyncWorkflowDraftHash(payload.hash, payload) + if (payload.hash) setSyncWorkflowDraftHash(payload.hash, payload, payload.client_save_id) onWorkflowDataUpdate?.(payload) payload.onApplied?.() diff --git a/web/app/components/workflow/store/workflow/workflow-draft-slice.ts b/web/app/components/workflow/store/workflow/workflow-draft-slice.ts index 752c412b59a..5d255760905 100644 --- a/web/app/components/workflow/store/workflow/workflow-draft-slice.ts +++ b/web/app/components/workflow/store/workflow/workflow-draft-slice.ts @@ -1,7 +1,9 @@ import type { Viewport } from 'reactflow' import type { StateCreator } from 'zustand' import type { Edge, EnvironmentVariable, Node } from '@/app/components/workflow/types' +import type { EnvironmentVariablePatch } from '@/service/workflow' import { debounce } from 'es-toolkit/compat' +import { v4 as uuidV4 } from 'uuid' type DebouncedFunc = { (fn: () => void): void @@ -9,11 +11,55 @@ type DebouncedFunc = { flush?: () => void } +type WorkflowDraftSaveInput = { + appId: string + generation: number + keepalive: boolean + environmentVariablePatch?: EnvironmentVariablePatch +} + +type WorkflowDraftSaveEntry = { + clientSaveId: string + sequence: number + environmentVariablePatch?: EnvironmentVariablePatch +} + +type WorkflowDraftSaveSubmission = { + clientSaveId: string + predecessorSaves: WorkflowDraftSaveEntry[] + environmentVariablePatch?: EnvironmentVariablePatch +} + +export type WorkflowDraftSave = { + isCurrent: () => boolean + start: () => WorkflowDraftSaveSubmission | null + acknowledge: () => void + reject: () => void +} + +type PendingWorkflowDraftSave = WorkflowDraftSaveInput & + WorkflowDraftSaveEntry & + WorkflowDraftSaveSubmission & { + superseded: boolean + dispatched: boolean + acknowledged: boolean + rejected: boolean + } + export type WorkflowDraftSliceShape = { workflowDraftSyncPhase: 'idle' | 'preparing' | 'builder' setWorkflowDraftSyncPhase: (phase: WorkflowDraftSliceShape['workflowDraftSyncPhase']) => void enqueueWorkflowDraftOperation: (operation: () => Promise) => Promise - lastSavedWorkflowDraft: { appId: string; generation: number; hash: string } | null + runWorkflowDraftSave: ( + input: WorkflowDraftSaveInput, + operation: (save: WorkflowDraftSave) => Promise, + ) => Promise + lastSavedWorkflowDraft: { + appId: string + generation: number + hash: string + clientSaveId: string | null + } | null setLastSavedWorkflowDraft: (saved: WorkflowDraftSliceShape['lastSavedWorkflowDraft']) => void backupDraft?: { nodes: Node[] @@ -25,8 +71,14 @@ export type WorkflowDraftSliceShape = { setBackupDraft: (backupDraft?: WorkflowDraftSliceShape['backupDraft']) => void debouncedSyncWorkflowDraft: DebouncedFunc syncWorkflowDraftHash: string + syncWorkflowDraftSaveId: string | null syncWorkflowDraftIsEmpty: boolean | undefined - setSyncWorkflowDraftHash: (hash: string, graph?: { nodes: unknown[]; edges: unknown[] }) => void + setSyncWorkflowDraftHash: ( + hash: string, + graph?: { nodes: unknown[]; edges: unknown[] }, + clientSaveId?: string | null, + ) => void + adoptWorkflowDraftSaveId: (appId: string, hash: string, clientSaveId?: string | null) => void workflowDraftGeneration: number workflowDraftLocalRevision: number workflowDraftSavedRevision: number @@ -46,8 +98,12 @@ export type WorkflowDraftSliceShape = { } export const createWorkflowDraftSlice: StateCreator = (set, get) => { - // One queue per mounted editor, shared by saves and authoritative reloads. + // Unload saves start immediately; every queued operation still waits for them. let draftQueue: Promise = Promise.resolve() + const immediateSaves = new Set>() + const pendingSaves = new Map() + const unacknowledgedSaves = new Map() + let nextSaveSequence = 0 // Create the debounced function and store it with access to cancel/flush methods const debouncedFn = debounce((syncWorkflowDraft) => { syncWorkflowDraft() @@ -57,24 +113,123 @@ export const createWorkflowDraftSlice: StateCreator = ( workflowDraftSyncPhase: 'idle', setWorkflowDraftSyncPhase: (workflowDraftSyncPhase) => set({ workflowDraftSyncPhase }), enqueueWorkflowDraftOperation: (operation) => { - const next = draftQueue.catch(() => undefined).then(operation) + const next = draftQueue + .catch(() => undefined) + .then(async () => { + while (immediateSaves.size > 0) await Promise.allSettled([...immediateSaves]) + return operation() + }) draftQueue = next return next }, + runWorkflowDraftSave: (input, operation) => { + const pending: PendingWorkflowDraftSave = { + ...input, + clientSaveId: uuidV4(), + sequence: nextSaveSequence++, + predecessorSaves: [], + superseded: false, + dispatched: false, + acknowledged: false, + rejected: false, + } + pendingSaves.set(pending.clientSaveId, pending) + const save: WorkflowDraftSave = { + isCurrent: () => !pending.superseded, + start: () => { + if (pending.superseded) return null + const previous = [ + ...unacknowledgedSaves.values(), + ...(pending.keepalive ? pendingSaves.values() : []), + ].filter( + (candidate) => + candidate.sequence < pending.sequence && + candidate.appId === pending.appId && + candidate.generation === pending.generation, + ) + if (previous.length > 0) { + const predecessors = new Map() + for (const candidate of previous) { + for (const entry of [...candidate.predecessorSaves, candidate]) { + predecessors.set(entry.clientSaveId, { + clientSaveId: entry.clientSaveId, + sequence: entry.sequence, + environmentVariablePatch: entry.environmentVariablePatch, + }) + } + } + pending.predecessorSaves = [...predecessors.values()].sort( + (a, b) => a.sequence - b.sequence, + ) + for (const candidate of previous) candidate.superseded = true + } + pending.dispatched = true + return { + clientSaveId: pending.clientSaveId, + predecessorSaves: pending.predecessorSaves, + environmentVariablePatch: pending.environmentVariablePatch, + } + }, + acknowledge: () => { + pending.acknowledged = true + for (const candidate of unacknowledgedSaves.values()) { + if ( + candidate.sequence < pending.sequence && + candidate.appId === pending.appId && + candidate.generation === pending.generation + ) + unacknowledgedSaves.delete(candidate.clientSaveId) + } + }, + reject: () => { + pending.rejected = true + pending.environmentVariablePatch = undefined + }, + } + const run = () => + operation(save).finally(() => { + pendingSaves.delete(pending.clientSaveId) + if ( + pending.dispatched && + !pending.acknowledged && + !pending.superseded && + (!pending.rejected || pending.predecessorSaves.length > 0) && + pending.generation === get().workflowDraftGeneration + ) + unacknowledgedSaves.set(pending.clientSaveId, pending) + }) + if (!input.keepalive) return get().enqueueWorkflowDraftOperation(run) + + const result = run() + immediateSaves.add(result) + return result.finally(() => immediateSaves.delete(result)) + }, lastSavedWorkflowDraft: null, setLastSavedWorkflowDraft: (lastSavedWorkflowDraft) => set({ lastSavedWorkflowDraft }), backupDraft: undefined, setBackupDraft: (backupDraft) => set(() => ({ backupDraft })), debouncedSyncWorkflowDraft: debouncedFn, syncWorkflowDraftHash: '', + syncWorkflowDraftSaveId: null, syncWorkflowDraftIsEmpty: undefined, - setSyncWorkflowDraftHash: (syncWorkflowDraftHash, graph) => + setSyncWorkflowDraftHash: (syncWorkflowDraftHash, graph, clientSaveId) => set(() => ({ syncWorkflowDraftHash, + syncWorkflowDraftSaveId: clientSaveId ?? null, syncWorkflowDraftIsEmpty: graph ? graph.nodes.length === 0 && graph.edges.length === 0 : undefined, })), + // An unrelated receipt must not authorize replaying unresolved local variable patches. + adoptWorkflowDraftSaveId: (appId, hash, clientSaveId) => { + const state = get() + if (hash !== state.syncWorkflowDraftHash) return + const hasUnresolvedSave = [...pendingSaves.values(), ...unacknowledgedSaves.values()].some( + (candidate) => + candidate.appId === appId && candidate.generation === state.workflowDraftGeneration, + ) + if (!hasUnresolvedSave) set({ syncWorkflowDraftSaveId: clientSaveId ?? null }) + }, workflowDraftGeneration: 0, workflowDraftLocalRevision: 0, workflowDraftSavedRevision: 0, @@ -99,6 +254,7 @@ export const createWorkflowDraftSlice: StateCreator = ( }, invalidateWorkflowDraftSync: () => { debouncedFn.cancel() + unacknowledgedSaves.clear() const workflowDraftGeneration = get().workflowDraftGeneration + 1 set({ workflowDraftGeneration, diff --git a/web/app/components/workflow/update-dsl-modal.tsx b/web/app/components/workflow/update-dsl-modal.tsx index 036e8f3f8d6..52b31e0b199 100644 --- a/web/app/components/workflow/update-dsl-modal.tsx +++ b/web/app/components/workflow/update-dsl-modal.tsx @@ -63,8 +63,14 @@ function UpdateDSLModal({ const handleWorkflowUpdate = useCallback( async (app_id: string) => { - const { graph, features, hash, conversation_variables, environment_variables } = - await fetchWorkflowDraft(`/apps/${app_id}/workflows/draft`) + const { + graph, + features, + hash, + client_save_id, + conversation_variables, + environment_variables, + } = await fetchWorkflowDraft(`/apps/${app_id}/workflows/draft`) const { nodes, edges, viewport } = graph const importedNodes = initialNodes(nodes, edges) @@ -83,6 +89,7 @@ function UpdateDSLModal({ viewport, features: normalizeWorkflowFeatures(features), hash, + client_save_id, conversation_variables: conversation_variables || [], environment_variables: environment_variables || [], }, diff --git a/web/types/workflow.ts b/web/types/workflow.ts index c414e16f189..68bcf9bfdc0 100644 --- a/web/types/workflow.ts +++ b/web/types/workflow.ts @@ -1,6 +1,7 @@ import type { WorkflowDraftVariableResponse, WorkflowDraftVariableWithoutValueResponse, + WorkflowResponse, } from '@dify/contracts/api/console/apps/types.gen' import type { RefObject } from 'react' import type { Viewport } from 'reactflow' @@ -127,7 +128,7 @@ type VersionEnvironment = { name: string } -export type FetchWorkflowDraftResponse = { +export type FetchWorkflowDraftResponse = Pick & { id: string graph: { nodes: Node[]