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.
This commit is contained in:
twwu committed 2026-11-03 16:16:39 +08:00
1 parent b54c2625b3
commit 488a8d9307
42 files changed
+2159 -127

No files matched your search

+61 -1
View File
@@ -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),
},
)
@@ -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")
+3
View File
@@ -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(
+13
View File
@@ -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 |
@@ -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
+43 -3
View File
@@ -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)
@@ -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),
}
+71 -2
View File
@@ -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,
@@ -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)
@@ -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"]
@@ -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):
@@ -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()
@@ -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)
@@ -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(
@@ -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")
@@ -1321,6 +1321,7 @@ export type DefaultBlockConfigResponse = {
}
export type WorkflowResponse = {
client_save_id?: string | null
conversation_variables: Array<WorkflowConversationVariableResponse>
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<PredecessorDraftSavePayload>
}
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
+13
View File
@@ -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(),
})
/**
@@ -195,6 +195,7 @@ export type DefaultBlockConfigResponse = {
}
export type WorkflowResponse = {
client_save_id?: string | null
conversation_variables: Array<WorkflowConversationVariableResponse>
created_at: number
created_by?: SimpleAccountResponse | null
+1
View File
@@ -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(),
@@ -57,6 +57,7 @@ export type SnippetWorkflowPaginationResponse = {
export type DefaultBlockConfigsResponse = Array<DefaultBlockConfigItemResponse>
export type SnippetWorkflowResponse = {
client_save_id?: string | null
conversation_variables: Array<WorkflowConversationVariableResponse>
created_at: number
created_by?: SimpleAccountResponse | null
@@ -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(),
@@ -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.
@@ -90,6 +90,7 @@ const events = {
cursors: new Set<Parameters<Manager['onCursorUpdate']>[0]>(),
ready: new Set<Parameters<Manager['onGraphReadyChange']>[0]>(),
applied: new Set<Parameters<Manager['onServerDraftApplied']>[0]>(),
baseline: new Set<Parameters<Manager['onDraftBaseline']>[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<Awaited<ReturnType<typeof api.saveDraft>>>()
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(
@@ -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
@@ -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<Omit<SyncDraftWorkflowResponse, 'result'>>
>()
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<Omit<SyncDraftWorkflowResponse, 'result'>>()
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<Omit<SyncDraftWorkflowResponse, 'result'>>()
const unloading = createDeferred<Omit<SyncDraftWorkflowResponse, 'result'>>()
const finalUnload = createDeferred<Omit<SyncDraftWorkflowResponse, 'result'>>()
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<Omit<SyncDraftWorkflowResponse, 'result'>>()
const unloading = createDeferred<Omit<SyncDraftWorkflowResponse, 'result'>>()
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)(
@@ -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,
@@ -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<Pick<ServerDraftUpdate, 'hash' | 'client_save_id'>>>(),
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)
@@ -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<FetchWorkflowDraftResponse['conversation_variables']>
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)
@@ -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<ReturnType<typeof getPostParams>>,
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<SyncDraftResult | null> => {
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<typeof performLocalSync>) =>
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<ReturnType<typeof getPostParams>>,
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,
],
@@ -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()
})
}
@@ -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 || [])
@@ -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(
@@ -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 () => {
@@ -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')] })
@@ -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 &&
@@ -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<WorkflowResponse, 'client_save_id'> & {
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<ServerDraftChange, 'revision' | 'hash' | 'updated_at'> & {
export type ServerDraftUpdate = Pick<
ServerDraftChange,
'revision' | 'hash' | 'updated_at' | 'client_save_id'
> & {
update: Uint8Array
}
@@ -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<WorkflowResponse, 'hash' | 'client_save_id'>
export type WorkflowSyncAcknowledgement =
| ({ success: true } & WorkflowSyncResult)
| { success: false; error?: string }
@@ -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 = {
+3 -3
View File
@@ -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<FetchWorkflowDraftResponse, 'client_save_id'> & {
nodes: Node[]
edges: Edge[]
viewport?: Viewport
@@ -319,7 +319,7 @@ export const Workflow: FC<WorkflowProps> = 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?.()
@@ -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: <T>(operation: () => Promise<T>) => Promise<T>
lastSavedWorkflowDraft: { appId: string; generation: number; hash: string } | null
runWorkflowDraftSave: <T>(
input: WorkflowDraftSaveInput,
operation: (save: WorkflowDraftSave) => Promise<T>,
) => Promise<T>
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<WorkflowDraftSliceShape> = (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<unknown> = Promise.resolve()
const immediateSaves = new Set<Promise<unknown>>()
const pendingSaves = new Map<string, PendingWorkflowDraftSave>()
const unacknowledgedSaves = new Map<string, PendingWorkflowDraftSave>()
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<WorkflowDraftSliceShape> = (
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<string, WorkflowDraftSaveEntry>()
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<WorkflowDraftSliceShape> = (
},
invalidateWorkflowDraftSync: () => {
debouncedFn.cancel()
unacknowledgedSaves.clear()
const workflowDraftGeneration = get().workflowDraftGeneration + 1
set({
workflowDraftGeneration,
@@ -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 || [],
},
+2 -1
View File
@@ -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<WorkflowResponse, 'client_save_id'> & {
id: string
graph: {
nodes: Node[]