fix: prevent hidden-tab collaboration leader from saving stale drafts (#38997)

Co-authored-by: hjlarry <hjlarry@163.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: yyh <yuanyouhuilyz@gmail.com>
Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
This commit is contained in:
Kevin9703 2026-07-20 14:58:01 +08:00 committed by GitHub
parent 30df1433cc
commit 6022939bf0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
35 changed files with 3025 additions and 216 deletions

View File

@ -105,6 +105,7 @@ class SyncDraftWorkflowPayload(BaseModel):
graph: dict[str, Any]
features: dict[str, Any]
hash: str | None = None
is_collaborative: bool = Field(default=False, alias="_is_collaborative")
environment_variables: list[dict[str, Any]] = Field(
default_factory=list,
)
@ -610,6 +611,7 @@ class DraftWorkflowApi(Resource):
environment_variables=environment_variables,
conversation_variables=conversation_variables,
session=db.session(),
graph_only=args["is_collaborative"],
)
except WorkflowHashNotEqualError:
raise DraftWorkflowNotSync()

View File

@ -98,6 +98,7 @@ def handle_collaboration_event(sid, data):
6. workflow_update
7. comments_update
8. node_panel_presence
9. graph_view_state (session reports tab visibility; drives leader election)
"""
return collaboration_service.relay_collaboration_event(sid, data)

View File

@ -21859,6 +21859,7 @@ The subscription constructor of the trigger provider
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| _is_collaborative | boolean | | No |
| conversation_variables | [ object ] | | No |
| environment_variables | [ object ] | | No |
| features | object | | Yes |

View File

@ -3,7 +3,10 @@ from __future__ import annotations
import json
from typing import NotRequired, TypedDict, override
from redis.lock import Lock
from extensions.ext_redis import redis_client
from extensions.redis_names import serialize_redis_name
SESSION_STATE_TTL_SECONDS = 3600
SERVER_HEARTBEAT_TTL_SECONDS = 90
@ -12,6 +15,31 @@ WORKFLOW_LEADER_PREFIX = "workflow_leader:"
WS_SID_MAP_PREFIX = "ws_sid_map:"
WS_SERVER_HEARTBEAT_PREFIX = "ws_server_heartbeat:"
WS_SERVER_SESSIONS_PREFIX = "ws_server_sessions:"
GRAPH_VIEW_STATE_LOCK_PREFIX = "workflow_graph_view_state_lock:"
GRAPH_VIEW_STATE_LOCK_TIMEOUT_SECONDS = 15
_UPDATE_SESSION_GRAPH_ACTIVE_LUA = """
local raw = redis.call('HGET', KEYS[1], ARGV[1])
if not raw then
return 0
end
local decoded_ok, session_info = pcall(cjson.decode, raw)
if not decoded_ok or type(session_info) ~= 'table' then
return 0
end
local incoming_sequence = tonumber(ARGV[3])
local current_sequence = tonumber(session_info.graph_active_sequence)
if current_sequence and incoming_sequence <= current_sequence then
return 0
end
session_info.graph_active = ARGV[2] == '1'
session_info.graph_active_sequence = incoming_sequence
redis.call('HSET', KEYS[1], ARGV[1], cjson.encode(session_info))
return 1
"""
class WorkflowSessionInfo(TypedDict):
@ -156,6 +184,55 @@ class WorkflowCollaborationRepository:
return users
def get_session_info(self, workflow_id: str, sid: str) -> WorkflowSessionInfo | None:
raw = self._redis.hget(self.workflow_key(workflow_id), sid)
value = self._decode(raw)
if not value:
return None
try:
session_info = json.loads(value)
except (TypeError, json.JSONDecodeError):
return None
if not isinstance(session_info, dict):
return None
if "user_id" not in session_info or "username" not in session_info or "sid" not in session_info:
return None
user: WorkflowSessionInfo = {
"user_id": str(session_info["user_id"]),
"username": str(session_info["username"]),
"avatar": session_info.get("avatar"),
"sid": str(session_info["sid"]),
"connected_at": int(session_info.get("connected_at") or 0),
}
if isinstance(session_info.get("server_id"), str):
user["server_id"] = session_info["server_id"]
if isinstance(session_info.get("graph_active"), bool):
user["graph_active"] = session_info["graph_active"]
return user
def update_session_graph_active(self, workflow_id: str, sid: str, active: bool, sequence: int) -> bool:
"""Atomically apply a graph visibility update when its client sequence is newer."""
# RedisClientWrapper prefixes regular hash calls, but eval is delegated to the raw client.
workflow_key = serialize_redis_name(self.workflow_key(workflow_id))
result = self._redis.eval(
_UPDATE_SESSION_GRAPH_ACTIVE_LUA,
1,
workflow_key,
sid,
"1" if active else "0",
sequence,
)
return bool(result)
def graph_view_state_lock(self, workflow_id: str) -> Lock:
"""Serialize visibility state changes and their leader-election side effects."""
return self._redis.lock(
f"{GRAPH_VIEW_STATE_LOCK_PREFIX}{workflow_id}",
timeout=GRAPH_VIEW_STATE_LOCK_TIMEOUT_SECONDS,
)
def refresh_server_heartbeat(self, server_id: str) -> None:
self._redis.set(self.server_key(server_id), "1", ex=SERVER_HEARTBEAT_TTL_SECONDS)

View File

@ -8,6 +8,7 @@ import uuid
from collections.abc import Mapping
from typing import Any, override
from socketio.exceptions import TimeoutError as SocketIOTimeoutError # type: ignore[reportMissingTypeStubs]
from sqlalchemy import select
from sqlalchemy.orm import Session
@ -18,6 +19,7 @@ from repositories.workflow_collaboration_repository import WorkflowCollaboration
logger = logging.getLogger(__name__)
SERVER_HEARTBEAT_INTERVAL_SECONDS = 30
SYNC_REQUEST_TIMEOUT_SECONDS = 15
_PROCESS_SERVER_ID: str | None = None
_PROCESS_SERVER_ID_PID: int | None = None
@ -38,7 +40,8 @@ class WorkflowCollaborationService:
Socket.IO rooms are process-local unless backed by a message queue, while online users and leader election live in
Redis. Each websocket worker writes a small heartbeat keyed by `server_id`; session rows store their owner so a
worker can distinguish a live remote sid from a stale sid left behind by a dead worker.
worker can distinguish a live remote sid from a stale sid left behind by a dead worker. Visibility events are
ordered per socket, and sync requests wait for the selected saver to acknowledge its draft write.
"""
_heartbeat_started: bool
@ -128,6 +131,9 @@ class WorkflowCollaborationService:
"sid": sid,
"connected_at": int(time.time()),
"server_id": self.server_id,
# Joins are assumed visible; hidden tabs re-report via a graph_view_state
# event right after receiving the post-join "status" emit.
"graph_active": True,
}
self._repository.set_session_info(workflow_id, session_info)
@ -158,7 +164,8 @@ class WorkflowCollaborationService:
self.handle_leader_disconnect(workflow_id, sid)
self.broadcast_online_users(workflow_id)
def relay_collaboration_event(self, sid: str, data: Mapping[str, object]) -> tuple[dict[str, str], int]:
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."""
mapping = self._repository.get_sid_mapping(sid)
if not mapping:
return {"msg": "unauthorized"}, 401
@ -174,11 +181,54 @@ class WorkflowCollaborationService:
if not event_type:
return {"msg": "invalid event type"}, 400
if event_type == "graph_view_state":
if not isinstance(event_data, Mapping):
return {"msg": "invalid graph_view_state"}, 400
graph_active = event_data.get("graphActive")
sequence = event_data.get("sequence")
if not isinstance(graph_active, bool):
return {"msg": "invalid graph_view_state"}, 400
if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 0:
return {"msg": "invalid graph_view_state"}, 400
# Sequence the visibility write together with its leader side effect. Otherwise a newer
# visible event can land after the Lua write but before an older hidden handler demotes.
with self._repository.graph_view_state_lock(workflow_id):
applied = self._repository.update_session_graph_active(workflow_id, sid, graph_active, sequence)
if not applied:
return {"msg": "graph_view_state_ignored"}, 200
# Write the flag before re-electing so tier-1 selection already excludes the
# session that just went hidden (including one _ensure_leader just promoted).
if not graph_active:
self._demote_leader_if_hidden(workflow_id, sid)
return {"msg": "graph_view_state_updated"}, 200
if event_type == "sync_request":
if not isinstance(event_data, Mapping):
return {"msg": "invalid sync_request"}, 400
request_id = event_data.get("requestId")
if not isinstance(request_id, str) or not request_id.strip():
return {"msg": "invalid sync_request"}, 400
leader_sid = self._repository.get_current_leader(workflow_id)
target_sid: str | None
if leader_sid and self.is_session_active(workflow_id, leader_sid):
target_sid = leader_sid
if self._is_session_graph_active(workflow_id, leader_sid):
target_sid = leader_sid
else:
# The leader is connected but its tab is hidden: its canvas is frozen
# (rAF paused), so saving through it would persist stale data. Hand
# leadership to a visible session — or, if every tab is hidden, to the
# requester, whose canvas at least contains its own edits.
replacement = self._select_graph_leader(workflow_id, preferred_sid=sid)
if replacement and replacement != leader_sid:
self._repository.set_leader(workflow_id, replacement)
self.broadcast_leader_change(workflow_id, replacement)
target_sid = replacement
else:
target_sid = leader_sid
else:
if leader_sid:
self._repository.delete_leader(workflow_id)
@ -188,15 +238,73 @@ class WorkflowCollaborationService:
self.broadcast_leader_change(workflow_id, target_sid)
if not target_sid:
return {"msg": "no_active_leader"}, 200
return {"msg": "no_active_leader", "requestId": request_id}, 503
target_data = dict(event_data)
target_data["requestId"] = request_id
try:
result = self._socketio.call(
"collaboration_update",
{"type": event_type, "userId": user_id, "data": target_data, "timestamp": timestamp},
to=target_sid,
timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
)
except SocketIOTimeoutError:
logger.warning(
"Workflow collaboration sync request timed out: workflow_id=%s requester_sid=%s target_sid=%s",
workflow_id,
sid,
target_sid,
)
return {"msg": "sync_request_timeout", "requestId": request_id}, 504
if not isinstance(result, Mapping) or result.get("success") is not True:
response: dict[str, object] = {
"msg": "workflow_sync_failed",
"requestId": request_id,
"success": False,
}
if isinstance(result, Mapping) and isinstance(result.get("error"), str):
response["error"] = result["error"]
return response, 502
workflow_hash = result.get("hash")
updated_at = result.get("updatedAt")
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
return {
"msg": "workflow_synced",
"requestId": request_id,
"success": True,
"hash": workflow_hash,
"updatedAt": updated_at,
}, 200
if event_type == "graph_resync_request":
leader_sid = self._repository.get_current_leader(workflow_id)
resync_target_sid: str | None
if leader_sid and self.is_session_active(workflow_id, leader_sid):
resync_target_sid = leader_sid
else:
if leader_sid:
self._repository.delete_leader(workflow_id)
resync_target_sid = self._select_graph_leader(workflow_id, preferred_sid=sid)
if resync_target_sid:
self._repository.set_leader(workflow_id, resync_target_sid)
self.broadcast_leader_change(workflow_id, resync_target_sid)
if not resync_target_sid:
return {"msg": "no_active_leader"}, 503
self._socketio.emit(
"collaboration_update",
{"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
room=target_sid,
to=resync_target_sid,
)
return {"msg": "sync_request_forwarded"}, 200
return {"msg": "graph_resync_request_forwarded"}, 200
self._socketio.emit(
"collaboration_update",
@ -329,17 +437,56 @@ class WorkflowCollaborationService:
self._repository.set_leader(workflow_id, sid)
self.broadcast_leader_change(workflow_id, sid)
def _select_graph_leader(self, workflow_id: str, preferred_sid: str | None = None) -> str | None:
session_sids = [
session["sid"]
def _select_graph_leader(
self,
workflow_id: str,
preferred_sid: str | None = None,
*,
require_graph_active: bool = False,
) -> str | None:
"""Pick a leader, preferring sessions whose canvas tab is visible.
Hidden tabs freeze rAF-driven CRDT->canvas application, so a visible session is
always the freshest saver. When every tab is hidden the room must still keep a
leader (followers drop sync_requests unless they believe they are the leader),
so tier 2 falls back to any active session unless require_graph_active is set.
"""
active_sessions = [
session
for session in self._repository.list_sessions(workflow_id)
if session.get("graph_active", True) and self.is_session_active(workflow_id, session["sid"])
if self.is_session_active(workflow_id, session["sid"])
]
if not session_sids:
visible_sids = [session["sid"] for session in active_sessions if session.get("graph_active", True)]
candidate_sids = visible_sids
if not candidate_sids and not require_graph_active:
candidate_sids = [session["sid"] for session in active_sessions]
if not candidate_sids:
return None
if preferred_sid and preferred_sid in session_sids:
if preferred_sid and preferred_sid in candidate_sids:
return preferred_sid
return session_sids[0]
return candidate_sids[0]
def _is_session_graph_active(self, workflow_id: str, sid: str) -> bool:
"""Default to True on missing/unreadable session data so read failures never churn leadership."""
session_info = self._repository.get_session_info(workflow_id, sid)
if session_info is None:
return True
return bool(session_info.get("graph_active", True))
def _demote_leader_if_hidden(self, workflow_id: str, hidden_sid: str) -> None:
current_leader = self._repository.get_current_leader(workflow_id)
if current_leader != hidden_sid:
return
new_leader = self._select_graph_leader(workflow_id, require_graph_active=True)
# No visible session: keep the hidden leader rather than leaving the room leaderless.
if not new_leader or new_leader == hidden_sid:
return
self._repository.set_leader(workflow_id, new_leader)
self.broadcast_leader_change(workflow_id, new_leader)
def is_session_active(self, workflow_id: str, sid: str) -> bool:
if not sid:

View File

@ -322,6 +322,7 @@ class WorkflowService:
session: Session,
commit: bool = True,
sync_agent_bindings: bool = True,
graph_only: bool = False,
) -> Workflow:
"""
Sync draft workflow.
@ -338,8 +339,10 @@ class WorkflowService:
if workflow and workflow.unique_hash != unique_hash:
raise WorkflowHashNotEqualError()
# validate features structure
self.validate_features_structure(app_model=app_model, features=features)
# Collaboration persists features and variables through dedicated endpoints. A graph save
# must not overwrite those newer database values with another collaborator's stale cache.
if not graph_only or not workflow:
self.validate_features_structure(app_model=app_model, features=features)
# validate graph structure
self.validate_graph_structure(graph=graph)
@ -361,11 +364,12 @@ class WorkflowService:
# update draft workflow if found
else:
workflow.graph = json.dumps(graph)
workflow.features = json.dumps(features)
workflow.updated_by = account.id
workflow.updated_at = naive_utc_now()
workflow.environment_variables = environment_variables
workflow.conversation_variables = conversation_variables
if not graph_only:
workflow.features = json.dumps(features)
workflow.environment_variables = environment_variables
workflow.conversation_variables = conversation_variables
from services.agent.workflow_publish_service import WorkflowAgentPublishService

View File

@ -1,5 +1,6 @@
FLASK_APP=app.py
FLASK_DEBUG=0
ENABLE_COLLABORATION_MODE=false
SECRET_KEY='uhySf6a3aZuvRNfAlcr47paOw9TRYBY6j8ZHXpVw1yx5RP27Yj3w2uvI'
CONSOLE_API_URL=http://127.0.0.1:5001

View File

@ -398,6 +398,12 @@ class TestWorkflowEndpoints:
def test_workflow_copy_payload(self):
payload = SyncDraftWorkflowPayload(graph={}, features={})
assert payload.graph == {}
assert payload.model_dump()["is_collaborative"] is False
def test_workflow_sync_payload_accepts_collaboration_marker(self):
payload = SyncDraftWorkflowPayload.model_validate({"graph": {}, "features": {}, "_is_collaborative": True})
assert payload.is_collaborative is True
assert payload.model_dump()["is_collaborative"] is True
def test_workflow_mode_query(self):
payload = AdvancedChatWorkflowRunPayload(inputs={}, query="hi")

View File

@ -48,6 +48,101 @@ class TestWorkflowCollaborationRepository:
}
]
def test_list_sessions_preserves_graph_active_false(self, mock_redis: Mock) -> None:
# Arrange
mock_redis.hgetall.return_value = {
b"sid-1": b'{"user_id":"u-1","username":"Jane","sid":"sid-1","connected_at":2,"graph_active":false}',
}
repository = WorkflowCollaborationRepository()
# Act
result = repository.list_sessions("wf-1")
# Assert
assert result[0]["graph_active"] is False
def test_get_session_info_returns_session_with_optional_fields(self, mock_redis: Mock) -> None:
# Arrange
mock_redis.hget.return_value = (
b'{"user_id":"u-1","username":"Jane","sid":"sid-1","connected_at":2,'
b'"server_id":"server-1","graph_active":false}'
)
repository = WorkflowCollaborationRepository()
# Act
result = repository.get_session_info("wf-1", "sid-1")
# Assert
assert result == {
"user_id": "u-1",
"username": "Jane",
"avatar": None,
"sid": "sid-1",
"connected_at": 2,
"server_id": "server-1",
"graph_active": False,
}
mock_redis.hget.assert_called_once_with("workflow_online_users:wf-1", "sid-1")
def test_get_session_info_returns_none_on_missing_or_bad_json(self, mock_redis: Mock) -> None:
repository = WorkflowCollaborationRepository()
mock_redis.hget.return_value = None
assert repository.get_session_info("wf-1", "sid-1") is None
mock_redis.hget.return_value = b"not-json"
assert repository.get_session_info("wf-1", "sid-1") is None
mock_redis.hget.return_value = b'{"username":"missing-required-keys"}'
assert repository.get_session_info("wf-1", "sid-1") is None
def test_update_session_graph_active_uses_atomic_lua_with_prefixed_key(
self, mock_redis: Mock, monkeypatch: pytest.MonkeyPatch
) -> None:
# Arrange
mock_redis.eval.return_value = 1
monkeypatch.setattr(repo_module, "serialize_redis_name", lambda key: f"dify-prefix:{key}")
repository = WorkflowCollaborationRepository()
# Act
result = repository.update_session_graph_active("wf-1", "sid-1", active=False, sequence=7)
# Assert
assert result is True
script, key_count, workflow_key, sid, active, sequence = mock_redis.eval.call_args.args
assert key_count == 1
assert workflow_key == "dify-prefix:workflow_online_users:wf-1"
assert sid == "sid-1"
assert active == "0"
assert sequence == 7
assert "incoming_sequence <= current_sequence" in script
assert "cjson.encode(session_info)" in script
def test_update_session_graph_active_returns_false_when_lua_ignores_update(self, mock_redis: Mock) -> None:
# Arrange
mock_redis.eval.return_value = 0
repository = WorkflowCollaborationRepository()
# Act
result = repository.update_session_graph_active("wf-1", "sid-1", active=True, sequence=3)
# Assert
assert result is False
assert mock_redis.eval.call_args.args[-2:] == ("1", 3)
def test_graph_view_state_lock_is_scoped_to_workflow(self, mock_redis: Mock) -> None:
expected_lock = Mock()
mock_redis.lock.return_value = expected_lock
repository = WorkflowCollaborationRepository()
lock = repository.graph_view_state_lock("wf-1")
assert lock is expected_lock
mock_redis.lock.assert_called_once_with(
"workflow_graph_view_state_lock:wf-1",
timeout=repo_module.GRAPH_VIEW_STATE_LOCK_TIMEOUT_SECONDS,
)
def test_set_session_info_persists_payload(self, mock_redis: Mock) -> None:
# Arrange
mock_redis.exists.return_value = True

View File

@ -1,16 +1,20 @@
import logging
from collections.abc import Iterator
from contextlib import contextmanager, nullcontext
from unittest.mock import Mock, patch
import pytest
from socketio.exceptions import TimeoutError as SocketIOTimeoutError
from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
from services.workflow_collaboration_service import WorkflowCollaborationService
from services.workflow_collaboration_service import SYNC_REQUEST_TIMEOUT_SECONDS, WorkflowCollaborationService
class TestWorkflowCollaborationService:
@pytest.fixture
def service(self) -> tuple[WorkflowCollaborationService, Mock, Mock]:
repository = Mock(spec=WorkflowCollaborationRepository)
repository.graph_view_state_lock.return_value = nullcontext()
socketio = Mock()
return WorkflowCollaborationService(repository, socketio, server_id="server-1"), repository, socketio
@ -159,13 +163,13 @@ class TestWorkflowCollaborationService:
assert result == ({"msg": "invalid event type"}, 400)
def test_relay_collaboration_event_sync_request_forwards_to_active_leader(
def test_relay_collaboration_event_graph_resync_request_forwards_to_active_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
payload = {"type": "graph_resync_request", "data": {"reason": "reconnect"}, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
@ -173,14 +177,560 @@ class TestWorkflowCollaborationService:
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "sync_request_forwarded"}, 200)
assert result == ({"msg": "graph_resync_request_forwarded"}, 200)
socketio.emit.assert_called_once_with(
"collaboration_update",
{"type": "sync_request", "userId": "u-1", "data": {"reason": "join"}, "timestamp": 123},
room="sid-leader",
{
"type": "graph_resync_request",
"userId": "u-1",
"data": {"reason": "reconnect"},
"timestamp": 123,
},
to="sid-leader",
)
repository.set_leader.assert_not_called()
def test_relay_collaboration_event_graph_resync_request_reelects_when_leader_is_inactive(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-old"
payload = {"type": "graph_resync_request", "data": None, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=False),
patch.object(collaboration_service, "_select_graph_leader", return_value="sid-1") as select_leader,
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_resync_request_forwarded"}, 200)
repository.delete_leader.assert_called_once_with("wf-1")
select_leader.assert_called_once_with("wf-1", preferred_sid="sid-1")
repository.set_leader.assert_called_once_with("wf-1", "sid-1")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-1")
socketio.emit.assert_called_once_with(
"collaboration_update",
{"type": "graph_resync_request", "userId": "u-1", "data": None, "timestamp": 123},
to="sid-1",
)
def test_relay_collaboration_event_graph_resync_request_returns_503_without_active_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-old"
payload = {"type": "graph_resync_request", "data": None, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=False),
patch.object(collaboration_service, "_select_graph_leader", return_value=None),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "no_active_leader"}, 503)
repository.delete_leader.assert_called_once_with("wf-1")
repository.set_leader.assert_not_called()
socketio.emit.assert_not_called()
@pytest.mark.parametrize(
("graph_active", "sequence"),
[
("false", 1),
(0, 1),
(None, 1),
(False, True),
(False, -1),
(False, "1"),
(False, None),
],
)
def test_relay_collaboration_event_rejects_invalid_graph_view_state(
self,
service: tuple[WorkflowCollaborationService, Mock, Mock],
graph_active: object,
sequence: object,
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
payload = {
"type": "graph_view_state",
"data": {"graphActive": graph_active, "sequence": sequence},
"timestamp": 123,
}
with patch.object(collaboration_service, "refresh_session_state"):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "invalid graph_view_state"}, 400)
repository.update_session_graph_active.assert_not_called()
def test_relay_collaboration_event_requires_sync_request_id(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
with patch.object(collaboration_service, "refresh_session_state"):
result = collaboration_service.relay_collaboration_event(
"sid-1", {"type": "sync_request", "data": {"requestId": " "}, "timestamp": 123}
)
assert result == ({"msg": "invalid sync_request"}, 400)
socketio.call.assert_not_called()
def test_relay_collaboration_event_sync_request_forwards_to_active_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": True,
}
socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456}
payload = {
"type": "sync_request",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == (
{
"msg": "workflow_synced",
"requestId": "req-1",
"success": True,
"hash": "hash-2",
"updatedAt": 456,
},
200,
)
socketio.call.assert_called_once_with(
"collaboration_update",
{
"type": "sync_request",
"userId": "u-1",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
},
to="sid-leader",
timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
)
repository.set_leader.assert_not_called()
def test_relay_collaboration_event_sync_request_returns_target_failure(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": True,
}
socketio.call.return_value = {"success": False, "error": "draft_workflow_not_sync"}
payload = {"type": "sync_request", "data": {"requestId": "req-1"}, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == (
{
"msg": "workflow_sync_failed",
"requestId": "req-1",
"success": False,
"error": "draft_workflow_not_sync",
},
502,
)
def test_relay_collaboration_event_sync_request_returns_timeout(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": True,
}
socketio.call.side_effect = SocketIOTimeoutError()
payload = {"type": "sync_request", "data": {"requestId": "req-1"}, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "sync_request_timeout", "requestId": "req-1"}, 504)
@pytest.mark.parametrize(
"target_result",
[
None,
{"success": "true", "hash": "hash-2", "updatedAt": 456},
{"success": True, "hash": "", "updatedAt": 456},
{"success": True, "hash": "hash-2", "updatedAt": True},
],
)
def test_relay_collaboration_event_sync_request_rejects_invalid_target_response(
self, service: tuple[WorkflowCollaborationService, Mock, Mock], target_result: object
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": True,
}
socketio.call.return_value = target_result
payload = {"type": "sync_request", "data": {"requestId": "req-1"}, "timestamp": 123}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result[1] == 502
def test_relay_collaboration_event_sync_request_reroutes_from_hidden_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
# Leader is connected but its tab is hidden; the visible requester takes over.
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": False,
}
repository.list_sessions.return_value = [
{
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": False,
},
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 2,
"graph_active": True,
},
]
socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456}
payload = {
"type": "sync_request",
"data": {"reason": "edit", "requestId": "req-1"},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result[1] == 200
assert result[0]["success"] is True
repository.set_leader.assert_called_once_with("wf-1", "sid-1")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-1")
socketio.call.assert_called_once_with(
"collaboration_update",
{
"type": "sync_request",
"userId": "u-1",
"data": {"reason": "edit", "requestId": "req-1"},
"timestamp": 123,
},
to="sid-1",
timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
)
def test_relay_collaboration_event_sync_request_all_hidden_falls_back_to_requester(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
# Every tab hidden: the requester still becomes leader so the save is not dropped.
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.get_session_info.return_value = {
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": False,
}
repository.list_sessions.return_value = [
{
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-leader",
"connected_at": 1,
"graph_active": False,
},
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 2,
"graph_active": False,
},
]
socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456}
payload = {
"type": "sync_request",
"data": {"reason": "edit", "requestId": "req-1"},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result[1] == 200
assert result[0]["success"] is True
repository.set_leader.assert_called_once_with("wf-1", "sid-1")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-1")
socketio.call.assert_called_once_with(
"collaboration_update",
{
"type": "sync_request",
"userId": "u-1",
"data": {"reason": "edit", "requestId": "req-1"},
"timestamp": 123,
},
to="sid-1",
timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
)
def test_relay_collaboration_event_graph_view_state_updates_without_broadcast(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-other"
repository.update_session_graph_active.return_value = True
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
"timestamp": 123,
}
with patch.object(collaboration_service, "refresh_session_state"):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
socketio.emit.assert_not_called()
repository.set_leader.assert_not_called()
def test_graph_view_state_serializes_visibility_write_with_leader_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
events: list[str] = []
@contextmanager
def graph_view_lock() -> Iterator[None]:
events.append("lock_enter")
yield
events.append("lock_exit")
repository.graph_view_state_lock.return_value = graph_view_lock()
repository.update_session_graph_active.side_effect = lambda *_args: events.append("visibility_write") or True
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 4},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(
collaboration_service,
"_demote_leader_if_hidden",
side_effect=lambda *_args: events.append("leader_demotion"),
),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
assert events == ["lock_enter", "visibility_write", "leader_demotion", "lock_exit"]
def test_relay_collaboration_event_graph_view_state_ignores_stale_update_without_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.update_session_graph_active.return_value = False
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 2},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "_demote_leader_if_hidden") as demote_leader,
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_ignored"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 2)
demote_leader.assert_not_called()
def test_relay_collaboration_event_graph_view_state_demotes_hidden_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-1"
repository.update_session_graph_active.return_value = True
repository.list_sessions.return_value = [
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
{
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-2",
"connected_at": 2,
"graph_active": True,
},
]
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
def test_relay_collaboration_event_graph_view_state_keeps_leader_when_all_hidden(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-1"
repository.update_session_graph_active.return_value = True
repository.list_sessions.return_value = [
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
]
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
patch.object(collaboration_service, "is_session_active", return_value=True),
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.set_leader.assert_not_called()
repository.delete_leader.assert_not_called()
broadcast_leader_change.assert_not_called()
def test_relay_collaboration_event_graph_view_state_non_leader_no_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-leader"
repository.update_session_graph_active.return_value = True
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
):
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
repository.set_leader.assert_not_called()
broadcast_leader_change.assert_not_called()
repository.list_sessions.assert_not_called()
def test_relay_collaboration_event_sync_request_reelects_active_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
@ -205,7 +755,12 @@ class TestWorkflowCollaborationService:
"graph_active": True,
},
]
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456}
payload = {
"type": "sync_request",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
}
def _is_session_active(_workflow_id: str, session_sid: str) -> bool:
return session_sid != "sid-old"
@ -217,14 +772,21 @@ class TestWorkflowCollaborationService:
):
result = collaboration_service.relay_collaboration_event("sid-2", payload)
assert result == ({"msg": "sync_request_forwarded"}, 200)
assert result[1] == 200
assert result[0]["success"] is True
repository.delete_leader.assert_called_once_with("wf-1")
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
socketio.emit.assert_called_once_with(
socketio.call.assert_called_once_with(
"collaboration_update",
{"type": "sync_request", "userId": "u-1", "data": {"reason": "join"}, "timestamp": 123},
room="sid-2",
{
"type": "sync_request",
"userId": "u-1",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
},
to="sid-2",
timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
)
def test_relay_collaboration_event_sync_request_returns_when_no_active_leader(
@ -234,7 +796,11 @@ class TestWorkflowCollaborationService:
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.get_current_leader.return_value = "sid-old"
repository.list_sessions.return_value = []
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
payload = {
"type": "sync_request",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
}
with (
patch.object(collaboration_service, "refresh_session_state"),
@ -242,9 +808,9 @@ class TestWorkflowCollaborationService:
):
result = collaboration_service.relay_collaboration_event("sid-2", payload)
assert result == ({"msg": "no_active_leader"}, 200)
assert result == ({"msg": "no_active_leader", "requestId": "req-1"}, 503)
repository.delete_leader.assert_called_once_with("wf-1")
socketio.emit.assert_not_called()
socketio.call.assert_not_called()
def test_relay_graph_event_unauthorized(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
# Arrange
@ -630,3 +1196,112 @@ class TestWorkflowCollaborationService:
socketio.manager.is_connected.return_value = False
assert collaboration_service.is_session_active("wf-1", "sid-remote") is False
@staticmethod
def _session(sid: str, connected_at: int, graph_active: bool) -> dict:
return {
"user_id": f"u-{sid}",
"username": sid,
"avatar": None,
"sid": sid,
"connected_at": connected_at,
"graph_active": graph_active,
}
def test_select_graph_leader_prefers_visible_preferred(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = [
self._session("sid-1", 1, graph_active=True),
self._session("sid-2", 2, graph_active=True),
]
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1", preferred_sid="sid-2")
assert result == "sid-2"
def test_select_graph_leader_visible_beats_hidden_preferred(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = [
self._session("sid-1", 1, graph_active=False),
self._session("sid-2", 2, graph_active=True),
]
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1", preferred_sid="sid-1")
assert result == "sid-2"
def test_select_graph_leader_all_hidden_falls_back_to_preferred(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = [
self._session("sid-1", 1, graph_active=False),
self._session("sid-2", 2, graph_active=False),
]
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1", preferred_sid="sid-2")
assert result == "sid-2"
def test_select_graph_leader_all_hidden_without_preference_picks_first(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = [
self._session("sid-1", 1, graph_active=False),
self._session("sid-2", 2, graph_active=False),
]
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1")
assert result == "sid-1"
def test_select_graph_leader_require_graph_active_returns_none_when_all_hidden(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = [
self._session("sid-1", 1, graph_active=False),
]
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1", require_graph_active=True)
assert result is None
def test_select_graph_leader_no_sessions_returns_none(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.list_sessions.return_value = []
with patch.object(collaboration_service, "is_session_active", return_value=True):
result = collaboration_service._select_graph_leader("wf-1")
assert result is None
def test_handle_leader_disconnect_elects_hidden_session_when_no_visible_remains(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
) -> None:
collaboration_service, repository, _socketio = service
repository.get_current_leader.return_value = "sid-old"
repository.list_sessions.return_value = [
self._session("sid-2", 2, graph_active=False),
]
with (
patch.object(collaboration_service, "is_session_active", return_value=True),
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
):
collaboration_service.handle_leader_disconnect("wf-1", "sid-old")
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")

View File

@ -449,6 +449,98 @@ class TestWorkflowService:
assert workflow.features_dict == features
assert workflow.updated_by == account.id
def test_sync_draft_workflow_graph_only_preserves_independently_updated_draft_fields(
self, workflow_service: WorkflowService, sqlite_session: Session
):
app = TestWorkflowAssociatedDataFactory.create_app()
account = TestWorkflowAssociatedDataFactory.create_account()
existing_features = {"file_upload": {"enabled": True}}
existing_environment_variables = [
StringVariable(id="env-new", name="region", value="new", selector=["env", "region"])
]
existing_conversation_variables = [
StringVariable(
id="conversation-new",
name="topic",
value="new",
selector=["conversation", "topic"],
)
]
workflow = TestWorkflowAssociatedDataFactory.create_workflow(
features=existing_features,
environment_variables=existing_environment_variables,
conversation_variables=existing_conversation_variables,
)
sqlite_session.add(workflow)
sqlite_session.commit()
unique_hash = workflow.unique_hash
updated_graph = TestWorkflowAssociatedDataFactory.create_valid_workflow_graph()
updated_graph["nodes"][0]["data"]["title"] = "Updated graph"
with patch("services.workflow_service.app_draft_workflow_was_synced"):
result = workflow_service.sync_draft_workflow(
app_model=app,
graph=updated_graph,
features={"file_upload": {"enabled": False}},
unique_hash=unique_hash,
account=account,
environment_variables=[
StringVariable(id="env-old", name="region", value="old", selector=["env", "region"])
],
conversation_variables=[
StringVariable(
id="conversation-old",
name="topic",
value="old",
selector=["conversation", "topic"],
)
],
session=sqlite_session,
graph_only=True,
)
sqlite_session.refresh(workflow)
assert result is workflow
assert workflow.graph_dict == updated_graph
assert workflow.features_dict == existing_features
assert workflow.environment_variables == existing_environment_variables
assert workflow.conversation_variables == existing_conversation_variables
def test_sync_draft_workflow_graph_only_creates_complete_initial_draft(
self, workflow_service: WorkflowService, sqlite_session: Session
):
app = TestWorkflowAssociatedDataFactory.create_app()
account = TestWorkflowAssociatedDataFactory.create_account()
graph = TestWorkflowAssociatedDataFactory.create_valid_workflow_graph()
features = {"file_upload": {"enabled": True}}
environment_variables = [StringVariable(id="env-id", name="region", value="new", selector=["env", "region"])]
conversation_variables = [
StringVariable(
id="conversation-id",
name="topic",
value="new",
selector=["conversation", "topic"],
)
]
with patch("services.workflow_service.app_draft_workflow_was_synced"):
workflow = workflow_service.sync_draft_workflow(
app_model=app,
graph=graph,
features=features,
unique_hash=None,
account=account,
environment_variables=environment_variables,
conversation_variables=conversation_variables,
session=sqlite_session,
graph_only=True,
)
assert workflow.graph_dict == graph
assert workflow.features_dict == features
assert workflow.environment_variables == environment_variables
assert workflow.conversation_variables == conversation_variables
def test_sync_draft_workflow_raises_hash_not_equal_error(
self, workflow_service: WorkflowService, sqlite_session: Session
):

View File

@ -4014,11 +4014,6 @@
"count": 1
}
},
"web/app/components/workflow-app/hooks/use-workflow-refresh-draft.ts": {
"typescript/no-explicit-any": {
"count": 2
}
},
"web/app/components/workflow-app/hooks/use-workflow-run-callbacks.ts": {
"no-restricted-imports": {
"count": 2

View File

@ -1003,6 +1003,7 @@ export type WorkflowResponse = {
}
export type SyncDraftWorkflowPayload = {
_is_collaborative?: boolean
conversation_variables?: Array<{
[key: string]: unknown
}>

View File

@ -655,6 +655,7 @@ export const zDefaultBlockConfigResponse = z.record(z.string(), z.unknown())
* SyncDraftWorkflowPayload
*/
export const zSyncDraftWorkflowPayload = z.object({
_is_collaborative: z.boolean().optional().default(false),
conversation_variables: z.array(z.record(z.string(), z.unknown())).optional(),
environment_variables: z.array(z.record(z.string(), z.unknown())).optional(),
features: z.record(z.string(), z.unknown()),

View File

@ -1,6 +1,6 @@
import type { ReactNode } from 'react'
import type { WorkflowProps } from '@/app/components/workflow'
import { fireEvent, screen, waitFor } from '@testing-library/react'
import { act, fireEvent, screen, waitFor } from '@testing-library/react'
import { useStore as useAppStore } from '@/app/components/app/store'
import { ChatVarType } from '@/app/components/workflow/panel/chat-variable-panel/type'
import { BlockEnum } from '@/app/components/workflow/types'
@ -15,6 +15,13 @@ const mockFetchWorkflowDraft = vi.hoisted(() => vi.fn())
const mockOnVarsAndFeaturesUpdate = vi.hoisted(() => vi.fn())
const mockOnWorkflowUpdate = vi.hoisted(() => vi.fn())
const mockOnSyncRequest = vi.hoisted(() => vi.fn())
const mockOnGraphReloadRequired = vi.hoisted(() => vi.fn())
const mockOnGraphReadyChange = vi.hoisted(() => vi.fn())
const mockRefreshGraphSynchronously = vi.hoisted(() => vi.fn())
const mockReplaceGraphFromReactFlow = vi.hoisted(() => vi.fn())
const mockCanPersistLocalGraph = vi.hoisted(() => vi.fn())
const mockIsGraphReloadCurrent = vi.hoisted(() => vi.fn())
const mockRetryGraphReload = vi.hoisted(() => vi.fn())
const hookFns = {
doSyncWorkflowDraft: vi.fn(),
@ -64,7 +71,16 @@ const collaborationRuntime = vi.hoisted(() => ({
const collaborationListeners = vi.hoisted(() => ({
varsAndFeaturesUpdate: null as null | ((update: unknown) => void | Promise<void>),
workflowUpdate: null as null | (() => void | Promise<void>),
syncRequest: null as null | (() => void),
syncRequest: null as
| null
| ((request: {
requestId: string
acknowledge: (result: { success: boolean; hash?: string; updatedAt?: number }) => void
}) => void),
graphReloadRequired: null as
| null
| ((request: { generation: number; token: number; attempt: number }) => void | Promise<void>),
graphReadyChange: null as null | ((isReady: boolean) => void),
}))
let capturedContextProps: Record<string, unknown> | null = null
@ -145,10 +161,30 @@ vi.mock('@/app/components/workflow/collaboration/core/collaboration-manager', ()
return vi.fn()
},
),
onSyncRequest: mockOnSyncRequest.mockImplementation((handler: () => void) => {
collaborationListeners.syncRequest = handler
return vi.fn()
}),
onSyncRequest: mockOnSyncRequest.mockImplementation(
(handler: typeof collaborationListeners.syncRequest) => {
collaborationListeners.syncRequest = handler
return vi.fn()
},
),
onGraphReloadRequired: mockOnGraphReloadRequired.mockImplementation(
(handler: typeof collaborationListeners.graphReloadRequired) => {
collaborationListeners.graphReloadRequired = handler
return vi.fn()
},
),
onGraphReadyChange: mockOnGraphReadyChange.mockImplementation(
(handler: typeof collaborationListeners.graphReadyChange) => {
collaborationListeners.graphReadyChange = handler
handler?.(true)
return vi.fn()
},
),
refreshGraphSynchronously: mockRefreshGraphSynchronously,
replaceGraphFromReactFlow: mockReplaceGraphFromReactFlow,
canPersistLocalGraph: mockCanPersistLocalGraph,
isGraphReloadCurrent: mockIsGraphReloadCurrent,
retryGraphReload: mockRetryGraphReload,
},
}))
@ -371,7 +407,14 @@ describe('WorkflowMain', () => {
collaborationListeners.varsAndFeaturesUpdate = null
collaborationListeners.workflowUpdate = null
collaborationListeners.syncRequest = null
collaborationListeners.graphReloadRequired = null
collaborationListeners.graphReadyChange = null
mockFetchWorkflowDraft.mockReset()
mockCanPersistLocalGraph.mockReturnValue(true)
mockIsGraphReloadCurrent.mockReturnValue(true)
mockReplaceGraphFromReactFlow.mockReturnValue(true)
hookFns.doSyncWorkflowDraft.mockResolvedValue({ hash: 'saved-hash', updatedAt: 2 })
hookFns.handleRefreshWorkflowDraft.mockResolvedValue(true)
useAppStore.setState({ appDetail: undefined })
})
@ -488,6 +531,20 @@ describe('WorkflowMain', () => {
expect(collaborationRuntime.stopCursorTracking).toHaveBeenCalled()
})
it('blocks canvas input until the collaborative graph is ready', () => {
collaborationRuntime.isEnabled = true
render(<WorkflowMain nodes={[]} edges={[]} viewport={{ x: 0, y: 0, zoom: 1 }} />)
expect(screen.queryByTestId('collaboration-graph-loading')).not.toBeInTheDocument()
act(() => collaborationListeners.graphReadyChange?.(false))
expect(screen.getByRole('status')).toHaveTextContent('workflow.common.syncingData')
act(() => collaborationListeners.graphReadyChange?.(true))
expect(screen.queryByRole('status')).not.toBeInTheDocument()
})
it('subscribes collaboration listeners and handles sync/workflow update callbacks', async () => {
collaborationRuntime.isEnabled = true
mockFetchWorkflowDraft.mockResolvedValue({
@ -510,8 +567,19 @@ describe('WorkflowMain', () => {
expect(mockOnWorkflowUpdate).toHaveBeenCalled()
expect(mockOnSyncRequest).toHaveBeenCalled()
collaborationListeners.syncRequest?.()
expect(hookFns.doSyncWorkflowDraft).toHaveBeenCalled()
const acknowledge = vi.fn()
collaborationListeners.syncRequest?.({ requestId: 'request-1', acknowledge })
expect(mockRefreshGraphSynchronously).toHaveBeenCalled()
await waitFor(() => {
expect(hookFns.doSyncWorkflowDraft).toHaveBeenCalledWith(false, undefined, {
forceLocal: true,
})
expect(acknowledge).toHaveBeenCalledWith({
success: true,
hash: 'saved-hash',
updatedAt: 2,
})
})
await collaborationListeners.varsAndFeaturesUpdate?.({})
await collaborationListeners.workflowUpdate?.()
@ -527,6 +595,55 @@ describe('WorkflowMain', () => {
})
})
it('reloads the HTTP draft before trusting a reconnected leader document', async () => {
collaborationRuntime.isEnabled = true
const request = { generation: 2, token: 1, attempt: 0 }
render(<WorkflowMain nodes={[]} edges={[]} viewport={{ x: 0, y: 0, zoom: 1 }} />)
await collaborationListeners.graphReloadRequired?.(request)
expect(hookFns.handleRefreshWorkflowDraft).toHaveBeenCalledWith(false, {
shouldApply: expect.any(Function),
})
expect(mockReplaceGraphFromReactFlow).toHaveBeenCalledWith(request)
})
it('rejects a directed save without importing an untrusted CRDT graph', () => {
collaborationRuntime.isEnabled = true
mockCanPersistLocalGraph.mockReturnValue(false)
render(<WorkflowMain nodes={[]} edges={[]} viewport={{ x: 0, y: 0, zoom: 1 }} />)
const acknowledge = vi.fn()
collaborationListeners.syncRequest?.({ requestId: 'request-untrusted', acknowledge })
expect(acknowledge).toHaveBeenCalledWith({
success: false,
error: 'Collaborative graph is not ready to save.',
})
expect(mockRefreshGraphSynchronously).not.toHaveBeenCalled()
expect(hookFns.doSyncWorkflowDraft).not.toHaveBeenCalled()
})
it('retries an authoritative graph reload after a transient fetch failure', async () => {
vi.useFakeTimers()
try {
collaborationRuntime.isEnabled = true
hookFns.handleRefreshWorkflowDraft.mockResolvedValue(false)
const request = { generation: 2, token: 1, attempt: 0 }
render(<WorkflowMain nodes={[]} edges={[]} viewport={{ x: 0, y: 0, zoom: 1 }} />)
await collaborationListeners.graphReloadRequired?.(request)
expect(mockRetryGraphReload).not.toHaveBeenCalled()
vi.advanceTimersByTime(1000)
expect(mockRetryGraphReload).toHaveBeenCalledWith(request)
} finally {
vi.useRealTimers()
}
})
it('restores a local start placeholder for empty collaboration workflow updates', async () => {
collaborationRuntime.isEnabled = true
mockFetchWorkflowDraft.mockResolvedValue({

View File

@ -269,6 +269,7 @@ describe('FeaturesTrigger', () => {
})
mockHandleCheckBeforePublish.mockResolvedValue(true)
mockUseNodesSyncDraft.mockReturnValue({ handleSyncWorkflowDraft: mockHandleSyncWorkflowDraft })
mockHandleSyncWorkflowDraft.mockResolvedValue({ hash: 'draft-hash', updatedAt: 1 })
mockUseFeatures.mockImplementation((selector: (state: Record<string, unknown>) => unknown) =>
selector({ features: { file: {} } }),
)
@ -472,6 +473,43 @@ describe('FeaturesTrigger', () => {
// Verifies publishing behavior across warnings, validation, and success.
describe('Publishing', () => {
it('should wait for the draft save barrier before publishing', async () => {
const user = userEvent.setup()
let resolveDraftSync: ((value: { hash: string; updatedAt: number }) => void) | undefined
mockHandleSyncWorkflowDraft.mockReturnValueOnce(
new Promise((resolve) => {
resolveDraftSync = resolve
}),
)
renderWithToast(<FeaturesTrigger />)
await user.click(screen.getByRole('button', { name: 'publisher-publish' }))
await waitFor(() => {
expect(mockHandleSyncWorkflowDraft).toHaveBeenCalledWith(true)
})
expect(mockPublishWorkflow).not.toHaveBeenCalled()
resolveDraftSync?.({ hash: 'saved-hash', updatedAt: 2 })
await waitFor(() => {
expect(mockPublishWorkflow).toHaveBeenCalled()
})
})
it('should not publish when the draft save barrier fails', async () => {
const user = userEvent.setup()
mockHandleSyncWorkflowDraft.mockResolvedValueOnce(null)
renderWithToast(<FeaturesTrigger />)
await user.click(screen.getByRole('button', { name: 'publisher-publish' }))
await waitFor(() => {
expect(mockHandleSyncWorkflowDraft).toHaveBeenCalledWith(true)
})
expect(mockPublishWorkflow).not.toHaveBeenCalled()
})
it('should notify error and reject publish when checklist has warning nodes', async () => {
// Arrange
const user = userEvent.setup()

View File

@ -167,6 +167,9 @@ const FeaturesTrigger = () => {
// Then perform the detailed validation
if (await handleCheckBeforePublish()) {
const draftSyncResult = await handleSyncWorkflowDraft(true)
if (!draftSyncResult) throw new Error('Workflow draft sync failed')
const res = await publishWorkflow({
url: publishParams?.url || `/apps/${appID}/workflows/publish`,
title: publishParams?.title || '',
@ -206,6 +209,7 @@ const FeaturesTrigger = () => {
[
needWarningNodes,
handleCheckBeforePublish,
handleSyncWorkflowDraft,
publishWorkflow,
appID,
t,
@ -222,7 +226,7 @@ const FeaturesTrigger = () => {
const onPublisherToggle = useCallback(
(state: boolean) => {
if (state) handleSyncWorkflowDraft(true)
if (state) void handleSyncWorkflowDraft(true)
},
[handleSyncWorkflowDraft],
)

View File

@ -5,7 +5,8 @@ import type { Shape as HooksStoreShape } from '@/app/components/workflow/hooks-s
import type { Edge, Node } from '@/app/components/workflow/types'
import type { FetchWorkflowDraftResponse } from '@/types/workflow'
import { useAtomValue } from 'jotai'
import { useCallback, useEffect, useMemo, useRef } from 'react'
import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
import { useTranslation } from 'react-i18next'
import { useReactFlow } from 'reactflow'
import { useStore as useAppStore } from '@/app/components/app/store'
import { useFeaturesStore } from '@/app/components/base/features/hooks'
@ -40,12 +41,20 @@ type WorkflowDataUpdatePayload = Pick<
FetchWorkflowDraftResponse,
'features' | 'conversation_variables' | 'environment_variables'
>
const GRAPH_RELOAD_RETRY_BASE_DELAY = 1000
const GRAPH_RELOAD_RETRY_MAX_DELAY = 30_000
const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
const { t } = useTranslation()
const featuresStore = useFeaturesStore()
const workflowStore = useWorkflowStore()
const appId = useStore((s) => s.appId)
const appDetail = useAppStore((s) => s.appDetail)
const containerRef = useRef<HTMLDivElement>(null)
const [collaborationGraphState, setCollaborationGraphState] = useState({
appId: null as string | null,
isReady: false,
})
const reactFlow = useReactFlow()
const { getWorkflowDraftGraphForCanvas } = useWorkflowDraftGraphForCanvas(appDetail?.mode)
@ -99,6 +108,14 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
}
}, [startCursorTracking, stopCursorTracking, reactFlow, isCollaborationEnabled])
useEffect(() => {
if (!appId || !isCollaborationEnabled) return
return collaborationManager.onGraphReadyChange((isReady) => {
setCollaborationGraphState({ appId, isReady })
})
}, [appId, isCollaborationEnabled])
const handleWorkflowDataUpdate = useCallback(
(payload: WorkflowDataUpdatePayload) => {
const { features, conversation_variables, environment_variables } = payload
@ -214,16 +231,70 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
isCollaborationEnabled,
])
// Listen for sync requests from other users (only processed by leader)
// The server directs this request to the selected saver. Do not gate it on the
// local leader flag because the preceding status event may still be in flight.
useEffect(() => {
if (!appId || !isCollaborationEnabled) return
const unsubscribe = collaborationManager.onSyncRequest(() => {
doSyncWorkflowDraft()
const unsubscribe = collaborationManager.onSyncRequest(({ acknowledge }) => {
if (!collaborationManager.canPersistLocalGraph()) {
acknowledge({ success: false, error: 'Collaborative graph is not ready to save.' })
return
}
collaborationManager.refreshGraphSynchronously()
void doSyncWorkflowDraft(false, undefined, { forceLocal: true })
.then((result) => {
acknowledge(
result
? { success: true, hash: result.hash, updatedAt: result.updatedAt }
: { success: false },
)
})
.catch(() => {
acknowledge({ success: false })
})
})
return unsubscribe
}, [appId, doSyncWorkflowDraft, isCollaborationEnabled])
useEffect(() => {
if (!appId || !isCollaborationEnabled) return
let retryTimer: ReturnType<typeof setTimeout> | undefined
let disposed = false
const unsubscribe = collaborationManager.onGraphReloadRequired(async (request) => {
if (retryTimer) {
clearTimeout(retryTimer)
retryTimer = undefined
}
const isCurrent = () => !disposed && collaborationManager.isGraphReloadCurrent(request)
const refreshed = await handleRefreshWorkflowDraft(false, { shouldApply: isCurrent })
if (!isCurrent()) return
if (refreshed) {
collaborationManager.replaceGraphFromReactFlow(request)
return
}
const retryDelay = Math.min(
GRAPH_RELOAD_RETRY_BASE_DELAY * 2 ** request.attempt,
GRAPH_RELOAD_RETRY_MAX_DELAY,
)
retryTimer = setTimeout(() => {
retryTimer = undefined
collaborationManager.retryGraphReload(request)
}, retryDelay)
})
return () => {
disposed = true
if (retryTimer) clearTimeout(retryTimer)
unsubscribe()
}
}, [appId, handleRefreshWorkflowDraft, isCollaborationEnabled])
const {
handleStartWorkflowRun,
handleWorkflowStartRunInChatflow,
@ -356,6 +427,25 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
>
<WorkflowChildren />
</WorkflowWithInnerContext>
{isCollaborationEnabled &&
(collaborationGraphState.appId !== appId || !collaborationGraphState.isReady) && (
<div
data-testid="collaboration-graph-loading"
className="absolute inset-0 z-50 flex cursor-wait items-center justify-center"
>
<div
role="status"
aria-live="polite"
className="flex items-center gap-1.5 rounded-lg border-[0.5px] border-components-panel-border bg-components-panel-bg-blur px-3 py-2 system-xs-medium text-text-secondary shadow-lg backdrop-blur-[5px]"
>
<span
aria-hidden="true"
className="i-ri-loader-4-line size-4 animate-spin text-text-accent motion-reduce:animate-none"
/>
<span>{t(($) => $['common.syncingData'], { ns: 'workflow' })}</span>
</div>
</div>
)}
</div>
)
}

View File

@ -11,7 +11,9 @@ const mockSetDraftUpdatedAt = vi.fn()
const mockGetNodesReadOnly = vi.fn()
const mockCollaborationIsConnected = vi.fn()
const mockCollaborationGetIsLeader = vi.fn()
const mockCollaborationEmitSyncRequest = vi.fn()
const mockCollaborationRequestWorkflowSync = vi.fn()
const mockCollaborationCanPersistLocalGraph = vi.fn()
const mockCollaborationCanFlushGraphOnPageClose = vi.fn()
let isCollaborationEnabled = false
let reactFlowState: {
@ -66,18 +68,13 @@ vi.mock('@/app/components/workflow/collaboration/core/collaboration-manager', ()
collaborationManager: {
isConnected: (...args: unknown[]) => mockCollaborationIsConnected(...args),
getIsLeader: (...args: unknown[]) => mockCollaborationGetIsLeader(...args),
emitSyncRequest: (...args: unknown[]) => mockCollaborationEmitSyncRequest(...args),
requestWorkflowSync: (...args: unknown[]) => mockCollaborationRequestWorkflowSync(...args),
canPersistLocalGraph: (...args: unknown[]) => mockCollaborationCanPersistLocalGraph(...args),
canFlushGraphOnPageClose: (...args: unknown[]) =>
mockCollaborationCanFlushGraphOnPageClose(...args),
},
}))
vi.mock('@/app/components/workflow/hooks/use-serial-async-callback', () => ({
useSerialAsyncCallback:
(fn: (...args: unknown[]) => Promise<void>, checkFn: () => boolean) =>
(...args: unknown[]) => {
if (!checkFn()) return fn(...args)
},
}))
const mockSyncWorkflowDraft = vi.fn()
vi.mock('@/service/workflow', () => ({
syncWorkflowDraft: (p: unknown) => mockSyncWorkflowDraft(p),
@ -140,6 +137,12 @@ describe('useNodesSyncDraft — handleRefreshWorkflowDraft(true) on 409', () =>
mockSyncWorkflowDraft.mockResolvedValue({ hash: 'new', updated_at: 1 })
mockCollaborationIsConnected.mockReturnValue(false)
mockCollaborationGetIsLeader.mockReturnValue(true)
mockCollaborationCanPersistLocalGraph.mockReturnValue(true)
mockCollaborationCanFlushGraphOnPageClose.mockReturnValue(true)
mockCollaborationRequestWorkflowSync.mockResolvedValue({
hash: 'remote-hash',
updatedAt: 2,
})
isCollaborationEnabled = false
})
@ -201,7 +204,7 @@ describe('useNodesSyncDraft — handleRefreshWorkflowDraft(true) on 409', () =>
const { result } = renderUseNodesSyncDraft()
await act(async () => {
await expect(result.current.doSyncWorkflowDraft(false, callbacks)).resolves.toBeUndefined()
await expect(result.current.doSyncWorkflowDraft(false, callbacks)).resolves.toBeNull()
})
expect(error.json).toHaveBeenCalled()
@ -465,7 +468,7 @@ describe('useNodesSyncDraft — handleRefreshWorkflowDraft(true) on 409', () =>
)
})
it('should emit sync request instead of syncing when current user is collaboration follower', async () => {
it('should wait for the leader save result when current user is collaboration follower', async () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(false)
@ -481,13 +484,102 @@ describe('useNodesSyncDraft — handleRefreshWorkflowDraft(true) on 409', () =>
await result.current.doSyncWorkflowDraft(false, callbacks)
})
expect(mockCollaborationEmitSyncRequest).toHaveBeenCalled()
expect(mockCollaborationRequestWorkflowSync).toHaveBeenCalled()
expect(mockSyncWorkflowDraft).not.toHaveBeenCalled()
expect(callbacks.onSuccess).not.toHaveBeenCalled()
expect(mockSetSyncWorkflowDraftHash).toHaveBeenCalledWith('remote-hash')
expect(mockSetDraftUpdatedAt).toHaveBeenCalledWith(2)
expect(callbacks.onSuccess).toHaveBeenCalled()
expect(callbacks.onError).not.toHaveBeenCalled()
expect(callbacks.onSettled).toHaveBeenCalled()
})
it('should report a failed leader save to the follower caller', async () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(false)
mockCollaborationRequestWorkflowSync.mockRejectedValue(new Error('sync timeout'))
const callbacks = {
onSuccess: vi.fn(),
onError: vi.fn(),
onSettled: vi.fn(),
}
const { result } = renderUseNodesSyncDraft()
let syncResult: unknown
await act(async () => {
syncResult = await result.current.doSyncWorkflowDraft(false, callbacks)
})
expect(syncResult).toBeNull()
expect(mockSyncWorkflowDraft).not.toHaveBeenCalled()
expect(callbacks.onSuccess).not.toHaveBeenCalled()
expect(callbacks.onError).toHaveBeenCalled()
expect(callbacks.onSettled).toHaveBeenCalled()
})
it('should force a directed sync request to save locally even before leader status arrives', async () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(false)
const { result } = renderUseNodesSyncDraft()
await act(async () => {
await result.current.doSyncWorkflowDraft(false, undefined, { forceLocal: true })
})
expect(mockCollaborationRequestWorkflowSync).not.toHaveBeenCalled()
expect(mockSyncWorkflowDraft).toHaveBeenCalled()
})
it('should not queue a directed local save behind the requester waiting for its ack', async () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(false)
let resolveRemoteSync: ((result: { hash: string; updatedAt: number }) => void) | undefined
mockCollaborationRequestWorkflowSync.mockReturnValueOnce(
new Promise((resolve) => {
resolveRemoteSync = resolve
}),
)
const { result } = renderUseNodesSyncDraft()
let requesterSync: Promise<unknown> | undefined
await act(async () => {
requesterSync = result.current.doSyncWorkflowDraft()
await Promise.resolve()
})
await act(async () => {
await result.current.doSyncWorkflowDraft(false, undefined, { forceLocal: true })
})
expect(mockSyncWorkflowDraft).toHaveBeenCalledTimes(1)
resolveRemoteSync?.({ hash: 'remote-hash', updatedAt: 2 })
await act(async () => {
await requesterSync
})
})
it('should not persist an untrusted collaborative graph', async () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(true)
mockCollaborationCanPersistLocalGraph.mockReturnValue(false)
const { result } = renderUseNodesSyncDraft()
let syncResult: unknown
await act(async () => {
syncResult = await result.current.doSyncWorkflowDraft(false)
})
expect(syncResult).toBeNull()
expect(mockSyncWorkflowDraft).not.toHaveBeenCalled()
})
it('should skip keepalive sync on page close when current user is collaboration follower', () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
@ -501,4 +593,19 @@ describe('useNodesSyncDraft — handleRefreshWorkflowDraft(true) on 409', () =>
expect(mockPostWithKeepalive).not.toHaveBeenCalled()
})
it('should allow the trusted sole leader to flush with keepalive while hidden', () => {
isCollaborationEnabled = true
mockCollaborationIsConnected.mockReturnValue(true)
mockCollaborationGetIsLeader.mockReturnValue(true)
mockCollaborationCanFlushGraphOnPageClose.mockReturnValue(true)
const { result } = renderUseNodesSyncDraft()
act(() => {
result.current.syncWorkflowDraftWhenPageClose()
})
expect(mockPostWithKeepalive).toHaveBeenCalledTimes(1)
})
})

View File

@ -78,9 +78,11 @@ describe('useWorkflowRefreshDraft — notUpdateCanvas parameter', () => {
it('should update canvas by default (notUpdateCanvas omitted)', async () => {
const { result } = renderHook(() => useWorkflowRefreshDraft())
let refreshed = false
await act(async () => {
result.current.handleRefreshWorkflowDraft()
refreshed = await result.current.handleRefreshWorkflowDraft()
})
expect(refreshed).toBe(true)
expect(mockHandleUpdateWorkflowCanvas).toHaveBeenCalledTimes(1)
})
@ -102,6 +104,78 @@ describe('useWorkflowRefreshDraft — notUpdateCanvas parameter', () => {
expect(mockHandleUpdateWorkflowCanvas).not.toHaveBeenCalled()
})
it('should discard a stale guarded response before it mutates the canvas or draft metadata', async () => {
let resolveFetch: ((value: typeof draftResponse) => void) | undefined
mockFetchWorkflowDraft.mockReturnValue(
new Promise((resolve) => {
resolveFetch = resolve
}),
)
let isCurrent = true
const { result } = renderHook(() => useWorkflowRefreshDraft())
let refreshPromise: Promise<boolean> | undefined
act(() => {
refreshPromise = result.current.handleRefreshWorkflowDraft(false, {
shouldApply: () => isCurrent,
})
})
isCurrent = false
await act(async () => {
resolveFetch?.(draftResponse)
await refreshPromise
})
await expect(refreshPromise).resolves.toBe(false)
expect(mockHandleUpdateWorkflowCanvas).not.toHaveBeenCalled()
expect(mockSetSyncWorkflowDraftHash).not.toHaveBeenCalled()
expect(mockSetEnvironmentVariables).not.toHaveBeenCalled()
expect(mockSetConversationVariables).not.toHaveBeenCalled()
})
it('keeps the syncing guard active until the newest overlapping refresh completes', async () => {
let resolveFirst: ((value: typeof draftResponse) => void) | undefined
let resolveSecond: ((value: typeof draftResponse) => void) | undefined
mockFetchWorkflowDraft
.mockReturnValueOnce(
new Promise((resolve) => {
resolveFirst = resolve
}),
)
.mockReturnValueOnce(
new Promise((resolve) => {
resolveSecond = resolve
}),
)
let firstIsCurrent = true
const { result } = renderHook(() => useWorkflowRefreshDraft())
let firstRefresh: Promise<boolean> | undefined
let secondRefresh: Promise<boolean> | undefined
act(() => {
firstRefresh = result.current.handleRefreshWorkflowDraft(false, {
shouldApply: () => firstIsCurrent,
})
secondRefresh = result.current.handleRefreshWorkflowDraft(false, {
shouldApply: () => true,
})
})
firstIsCurrent = false
mockSetIsSyncingWorkflowDraft.mockClear()
await act(async () => {
resolveFirst?.(draftResponse)
await firstRefresh
})
expect(mockSetIsSyncingWorkflowDraft).not.toHaveBeenCalledWith(false)
await act(async () => {
resolveSecond?.(draftResponse)
await secondRefresh
})
expect(mockSetIsSyncingWorkflowDraft).toHaveBeenLastCalledWith(false)
})
it('should still update hash even when notUpdateCanvas=true', async () => {
const { result } = renderHook(() => useWorkflowRefreshDraft())
await act(async () => {
@ -223,10 +297,13 @@ describe('useWorkflowRefreshDraft — notUpdateCanvas parameter', () => {
const { result } = renderHook(() => useWorkflowRefreshDraft())
act(() => {
result.current.handleRefreshWorkflowDraft()
let refreshed = true
await act(async () => {
refreshed = await result.current.handleRefreshWorkflowDraft()
})
expect(refreshed).toBe(false)
await waitFor(() => {
expect(mockSetIsWorkflowDataLoaded).toHaveBeenNthCalledWith(1, false)
expect(mockSetIsWorkflowDataLoaded).toHaveBeenNthCalledWith(2, true)

View File

@ -1,4 +1,8 @@
import type { SyncDraftCallback } from '@/app/components/workflow/hooks-store'
import type {
SyncDraftCallback,
SyncDraftOptions,
SyncDraftResult,
} from '@/app/components/workflow/hooks-store'
import type { WorkflowDraftFeaturesPayload } from '@/service/workflow'
import { useSuspenseQuery } from '@tanstack/react-query'
import { produce } from 'immer'
@ -125,6 +129,8 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => {
const syncWorkflowDraftWhenPageClose = useCallback(() => {
if (getNodesReadOnly()) return
if (isCollaborationEnabled && !collaborationManager.canFlushGraphOnPageClose()) return
const isFollower =
isCollaborationEnabled &&
collaborationManager.isConnected() &&
@ -137,23 +143,25 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => {
if (postParams) postWithKeepalive(`${API_PREFIX}${postParams.url}`, postParams.params)
}, [getPostParams, getNodesReadOnly, isCollaborationEnabled])
const performSync = useCallback(
async (notRefreshWhenSyncError?: boolean, callback?: SyncDraftCallback) => {
if (getNodesReadOnly()) return
const performLocalSync = useCallback(
async (
notRefreshWhenSyncError?: boolean,
callback?: SyncDraftCallback,
): Promise<SyncDraftResult | null> => {
if (getNodesReadOnly()) return null
const isFollower =
isCollaborationEnabled &&
collaborationManager.isConnected() &&
!collaborationManager.getIsLeader()
if (isFollower) {
collaborationManager.emitSyncRequest()
if (isCollaborationEnabled && !collaborationManager.canPersistLocalGraph()) {
callback?.onError?.()
callback?.onSettled?.()
return
return null
}
const baseParams = getPostParams()
if (!baseParams) return
if (!baseParams) {
callback?.onError?.()
callback?.onSettled?.()
return null
}
const { setSyncWorkflowDraftHash, setDraftUpdatedAt } = workflowStore.getState()
@ -172,6 +180,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => {
setSyncWorkflowDraftHash(res.hash)
setDraftUpdatedAt(res.updated_at)
callback?.onSuccess?.()
return { hash: res.hash, updatedAt: res.updated_at }
} catch (error: unknown) {
const responseError = error as {
bodyUsed?: boolean
@ -187,6 +196,7 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => {
}
}
callback?.onError?.()
return null
} finally {
callback?.onSettled?.()
}
@ -200,7 +210,39 @@ const useNodesSyncDraftBase = (getNodesReadOnly: () => boolean) => {
],
)
const doSyncWorkflowDraft = useSerialAsyncCallback(performSync, getNodesReadOnly)
const doSyncWorkflowDraftLocally = useSerialAsyncCallback(performLocalSync, getNodesReadOnly)
const doSyncWorkflowDraft = useCallback(
async (
notRefreshWhenSyncError?: boolean,
callback?: SyncDraftCallback,
options?: SyncDraftOptions,
): Promise<SyncDraftResult | null> => {
if (getNodesReadOnly()) return null
const shouldRequestLeader =
isCollaborationEnabled &&
collaborationManager.isConnected() &&
!collaborationManager.getIsLeader() &&
!options?.forceLocal
if (!shouldRequestLeader) return doSyncWorkflowDraftLocally(notRefreshWhenSyncError, callback)
try {
const result = await collaborationManager.requestWorkflowSync()
const { setSyncWorkflowDraftHash, setDraftUpdatedAt } = workflowStore.getState()
setSyncWorkflowDraftHash(result.hash)
setDraftUpdatedAt(result.updatedAt)
callback?.onSuccess?.()
return result
} catch {
callback?.onError?.()
return null
} finally {
callback?.onSettled?.()
}
},
[doSyncWorkflowDraftLocally, getNodesReadOnly, isCollaborationEnabled, workflowStore],
)
return {
doSyncWorkflowDraft,

View File

@ -1,18 +1,25 @@
import { useCallback } from 'react'
import { useCallback, useRef } from 'react'
import { useStore as useAppStore } from '@/app/components/app/store'
import { useWorkflowUpdate } from '@/app/components/workflow/hooks'
import { useWorkflowStore } from '@/app/components/workflow/store'
import { fetchWorkflowDraft } from '@/service/workflow'
import { useWorkflowDraftGraphForCanvas } from './use-workflow-draft-graph-for-canvas'
type RefreshWorkflowDraftOptions = {
shouldApply?: () => boolean
}
export const useWorkflowRefreshDraft = () => {
const appDetail = useAppStore((s) => s.appDetail)
const workflowStore = useWorkflowStore()
const refreshSequenceRef = useRef(0)
const { handleUpdateWorkflowCanvas } = useWorkflowUpdate()
const { getWorkflowDraftGraphForCanvas } = useWorkflowDraftGraphForCanvas(appDetail?.mode)
const handleRefreshWorkflowDraft = useCallback(
(notUpdateCanvas?: boolean) => {
(notUpdateCanvas?: boolean, options?: RefreshWorkflowDraftOptions) => {
if (options?.shouldApply && !options.shouldApply()) return Promise.resolve(false)
const {
appId,
setSyncWorkflowDraftHash,
@ -25,17 +32,16 @@ export const useWorkflowRefreshDraft = () => {
debouncedSyncWorkflowDraft,
} = workflowStore.getState()
if (
debouncedSyncWorkflowDraft &&
typeof (debouncedSyncWorkflowDraft as any).cancel === 'function'
)
(debouncedSyncWorkflowDraft as any).cancel()
debouncedSyncWorkflowDraft?.cancel?.()
const wasLoaded = isWorkflowDataLoaded
if (wasLoaded) setIsWorkflowDataLoaded(false)
if (wasLoaded && !options?.shouldApply) setIsWorkflowDataLoaded(false)
const refreshSequence = ++refreshSequenceRef.current
setIsSyncingWorkflowDraft(true)
fetchWorkflowDraft(`/apps/${appId}/workflows/draft`)
return fetchWorkflowDraft(`/apps/${appId}/workflows/draft`)
.then((response) => {
if (options?.shouldApply && !options.shouldApply()) return false
// Ensure we have a valid workflow structure with viewport
if (!notUpdateCanvas)
handleUpdateWorkflowCanvas(getWorkflowDraftGraphForCanvas(response.graph))
@ -58,12 +64,14 @@ export const useWorkflowRefreshDraft = () => {
)
setConversationVariables(response.conversation_variables || [])
setIsWorkflowDataLoaded(true)
return true
})
.catch(() => {
if (wasLoaded) setIsWorkflowDataLoaded(true)
if (wasLoaded && !options?.shouldApply) setIsWorkflowDataLoaded(true)
return false
})
.finally(() => {
setIsSyncingWorkflowDraft(false)
if (refreshSequence === refreshSequenceRef.current) setIsSyncingWorkflowDraft(false)
})
},
[getWorkflowDraftGraphForCanvas, handleUpdateWorkflowCanvas, workflowStore],

View File

@ -34,6 +34,7 @@ type CollaborationManagerInternals = {
leaderId: string | null
isLeader: boolean
graphViewActive: boolean | null
crdtTrusted: boolean
pendingInitialSync: boolean
onlineUsers: OnlineUser[]
graphImportLogs: unknown[]
@ -74,6 +75,7 @@ const setupManagerWithDoc = () => {
internals.doc = doc
internals.nodesMap = doc.getMap('nodes')
internals.edgesMap = doc.getMap('edges')
internals.crdtTrusted = true
return { manager, internals }
}

View File

@ -1,10 +1,12 @@
import type { Socket } from 'socket.io-client'
import type {
CollaborationUpdate,
GraphReloadRequest,
NodePanelPresenceMap,
OnlineUser,
RestoreCompleteData,
RestoreIntentData,
WorkflowSyncRequest,
} from '../../types/collaboration'
import type { Edge, Node } from '@/app/components/workflow/types'
import { LoroDoc, LoroMap } from 'loro-crdt'
@ -58,7 +60,17 @@ type CollaborationManagerInternals = {
leaderId: string | null
pendingInitialSync: boolean
pendingGraphImportEmit: boolean
pendingGraphResyncBroadcast: boolean
rejoinInProgress: boolean
graphViewActive: boolean | null
graphViewSequence: number
visibilityListenerAttached: boolean
crdtTrusted: boolean
rebuildCrdtOnNextConnect: boolean
reconnectedWithFreshDoc: boolean
awaitingSnapshotImport: boolean
graphReloadRequired: boolean
crdtGeneration: number
onlineUsers: OnlineUser[]
nodePanelPresence: NodePanelPresenceMap
cursors: Record<string, { x: number; y: number; userId: string; timestamp: number }>
@ -69,7 +81,7 @@ type CollaborationManagerInternals = {
setupSocketEventListeners: (socket: Socket) => void
setupSubscriptions: () => void
scheduleGraphImportEmit: () => void
emitGraphResyncRequest: () => void
emitGraphResyncRequest: () => boolean
broadcastCurrentGraph: () => void
requestInitialSyncIfNeeded: () => void
cleanupNodePanelPresence: (activeClientIds: Set<string>) => void
@ -140,6 +152,7 @@ const setupManagerWithDoc = () => {
internals.doc = doc
internals.nodesMap = doc.getMap('nodes')
internals.edgesMap = doc.getMap('edges')
internals.crdtTrusted = true
return { manager, internals }
}
@ -148,7 +161,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
vi.clearAllMocks()
})
it('emits cursor/sync/workflow events via collaboration_event when connected', () => {
it('emits cursor/sync/workflow events via collaboration_event when connected', async () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-connected')
@ -157,7 +170,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
manager.emitCursorMove({ x: 11, y: 22, userId: 'u-1', timestamp: Date.now() })
manager.emitSyncRequest()
const syncPromise = manager.requestWorkflowSync()
manager.emitWorkflowUpdate('wf-1')
expect(socket.emit).toHaveBeenCalledTimes(3)
@ -170,7 +183,156 @@ describe('CollaborationManager socket and subscription behavior', () => {
'workflow_update',
])
expect(payloads[0]?.data).toMatchObject({ x: 11, y: 22 })
expect(payloads[1]?.data.graphSnapshot).toBeInstanceOf(Uint8Array)
expect(payloads[2]?.data).toMatchObject({ appId: 'wf-1' })
const syncCall = socket.emit.mock.calls.find(
(call) => (call[1] as { type?: string })?.type === 'sync_request',
)
const acknowledge = syncCall?.[2] as ((...args: unknown[]) => void) | undefined
acknowledge?.({ success: true, hash: 'hash-2', updatedAt: 2 }, 200)
await expect(syncPromise).resolves.toEqual({ hash: 'hash-2', updatedAt: 2 })
})
it('rejects requestWorkflowSync when the server acknowledgement reports failure', async () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-sync-failure')
internals.currentAppId = 'app-1'
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
const syncPromise = manager.requestWorkflowSync()
const acknowledge = socket.emit.mock.calls[0]?.[2] as ((...args: unknown[]) => void) | undefined
acknowledge?.({ success: false, error: 'save failed' }, 502)
await expect(syncPromise).rejects.toThrow('save failed')
})
it('handles a directed sync request and acknowledges it even when local leader state is stale', () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-directed-sync')
const socketAcknowledgement = vi.fn()
let receivedRequest: WorkflowSyncRequest | undefined
internals.isLeader = false
internals.setupSocketEventListeners(socket as unknown as Socket)
manager.onSyncRequest((request) => {
receivedRequest = request
})
socket.trigger(
'collaboration_update',
{
type: 'sync_request',
userId: 'user-1',
data: {
requestId: 'request-1',
graphSnapshot: internals.doc!.export({ mode: 'snapshot' }),
},
timestamp: 1,
} satisfies CollaborationUpdate,
socketAcknowledgement,
)
expect(receivedRequest?.requestId).toBe('request-1')
receivedRequest?.acknowledge({ success: true, hash: 'hash-1', updatedAt: 1 })
receivedRequest?.acknowledge({ success: false, error: 'duplicate' })
expect(socketAcknowledgement).toHaveBeenCalledTimes(1)
expect(socketAcknowledgement).toHaveBeenCalledWith({
success: true,
hash: 'hash-1',
updatedAt: 1,
})
})
it('rejects a directed workflow sync without a valid graph snapshot', () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-invalid-sync-snapshot')
const socketAcknowledgement = vi.fn()
const syncRequestHandler = vi.fn()
internals.setupSocketEventListeners(socket as unknown as Socket)
manager.onSyncRequest(syncRequestHandler)
socket.trigger(
'collaboration_update',
{
type: 'sync_request',
userId: 'requester-user',
data: { requestId: 'request-without-snapshot' },
timestamp: 10,
} satisfies CollaborationUpdate,
socketAcknowledgement,
)
expect(syncRequestHandler).not.toHaveBeenCalled()
expect(socketAcknowledgement).toHaveBeenCalledWith({
success: false,
error: 'Collaborative graph is not ready to save.',
})
})
it('imports the requester snapshot before handling a directed workflow sync', () => {
const { manager, internals } = setupManagerWithDoc()
const { manager: requester, internals: requesterInternals } = setupManagerWithDoc()
const socket = createMockSocket('socket-directed-snapshot')
const socketAcknowledgement = vi.fn()
const requesterNode = createNode('requester-node', 'Requester edit')
requester.setNodes([], [requesterNode])
internals.setupSocketEventListeners(socket as unknown as Socket)
manager.onSyncRequest(({ acknowledge }) => {
expect(manager.getNodes()).toEqual([
expect.objectContaining({
id: requesterNode.id,
data: expect.objectContaining({ title: 'Requester edit' }),
}),
])
acknowledge({ success: true, hash: 'merged-hash', updatedAt: 10 })
})
socket.trigger(
'collaboration_update',
{
type: 'sync_request',
userId: 'requester-user',
data: {
requestId: 'request-with-snapshot',
graphSnapshot: requesterInternals.doc!.export({ mode: 'snapshot' }),
},
timestamp: 10,
} satisfies CollaborationUpdate,
socketAcknowledgement,
)
expect(socketAcknowledgement).toHaveBeenCalledWith({
success: true,
hash: 'merged-hash',
updatedAt: 10,
})
})
it('allows a hidden page-close flush only for the sole trusted leader session', () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-page-close')
internals.currentAppId = 'app-page-close'
internals.isLeader = true
internals.graphViewActive = false
internals.onlineUsers = [
{ user_id: 'user-1', username: 'User 1', avatar: '', sid: 'socket-page-close' },
]
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
expect(manager.canFlushGraphOnPageClose()).toBe(true)
internals.onlineUsers.push({
user_id: 'user-2',
username: 'User 2',
avatar: '',
sid: 'socket-other',
})
expect(manager.canFlushGraphOnPageClose()).toBe(false)
internals.graphViewActive = true
expect(manager.canFlushGraphOnPageClose()).toBe(true)
})
it('tries to rejoin on unauthorized and forces disconnect on unauthorized ack', () => {
@ -212,7 +374,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
const broadcastSpy = vi
.spyOn(internals, 'broadcastCurrentGraph')
.mockImplementation(() => undefined)
internals.isLeader = true
internals.isLeader = false
internals.setupSocketEventListeners(socket as unknown as Socket)
const varsFeatureHandler = vi.fn()
@ -306,7 +468,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
socket.trigger('collaboration_update', {
...baseUpdate,
type: 'sync_request',
data: {},
data: { graphSnapshot: internals.doc!.export({ mode: 'snapshot' }) },
} satisfies CollaborationUpdate)
socket.trigger('collaboration_update', {
...baseUpdate,
@ -370,7 +532,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
const emitGraphResyncRequestSpy = vi
.spyOn(internals, 'emitGraphResyncRequest')
.mockImplementation(() => undefined)
.mockReturnValue(true)
internals.cursors = {
stale: {
@ -437,6 +599,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
internals.pendingInitialSync = true
internals.isLeader = false
internals.crdtTrusted = false
socket.trigger('status', { isLeader: false })
expect(emitGraphResyncRequestSpy).toHaveBeenCalledTimes(1)
expect(internals.pendingInitialSync).toBe(false)
@ -676,11 +839,18 @@ describe('CollaborationManager socket and subscription behavior', () => {
internals.nodesMap = doc.getMap('nodes')
internals.edgesMap = doc.getMap('edges')
internals.broadcastCurrentGraph()
expect(sendGraphEventSpy).not.toHaveBeenCalled()
expect(sendGraphEventSpy).toHaveBeenCalledTimes(1)
manager.setNodes([], [createNode('n-broadcast')])
internals.broadcastCurrentGraph()
expect(sendGraphEventSpy).toHaveBeenCalledTimes(1)
expect(sendGraphEventSpy).toHaveBeenCalledTimes(2)
internals.crdtTrusted = false
manager.setNodes(manager.getNodes(), [...manager.getNodes(), createNode('n-stale')])
internals.broadcastCurrentGraph()
expect(manager.getNodes().some((node) => node.id === 'n-stale')).toBe(false)
expect(sendGraphEventSpy).toHaveBeenCalledTimes(2)
expect(internals.pendingGraphResyncBroadcast).toBe(true)
})
it('covers connect lifecycle branches including reconnect and force disconnect cleanup', async () => {
@ -731,6 +901,257 @@ describe('CollaborationManager socket and subscription behavior', () => {
expect(internals.activeConnections.size).toBe(0)
})
it('rebuilds the CRDT document on reconnect and reloads a re-elected leader from HTTP', async () => {
const manager = new CollaborationManager()
const internals = getManagerInternals(manager)
const socket = createMockSocket('socket-reconnect')
const authoritativeNode = createNode('node-latest', 'Latest')
const reactFlowStore: ReactFlowStore = {
getState: () => ({
getNodes: () => [authoritativeNode],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
}
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
const connectionId = await manager.connect('app-reconnect', reactFlowStore)
const firstDocument = internals.doc
socket.trigger('status', { isLeader: true })
expect(manager.canPersistLocalGraph()).toBe(true)
socket.trigger('disconnect', 'transport close')
expect(manager.canPersistLocalGraph()).toBe(false)
socket.trigger('connect')
expect(internals.doc).not.toBe(firstDocument)
expect(manager.getNodes()).toEqual([])
socket.trigger('status', { isLeader: true })
const reloadRequired = vi.fn()
const unsubscribe = manager.onGraphReloadRequired(reloadRequired)
expect(reloadRequired).toHaveBeenCalledTimes(1)
expect(manager.canPersistLocalGraph()).toBe(false)
const reloadRequest = reloadRequired.mock.calls[0]?.[0] as GraphReloadRequest
expect(manager.isGraphReloadCurrent(reloadRequest)).toBe(true)
internals.pendingGraphResyncBroadcast = true
const sendGraphEventSpy = vi
.spyOn(
manager as unknown as { sendGraphEvent: (payload: Uint8Array) => void },
'sendGraphEvent',
)
.mockImplementation(() => undefined)
manager.replaceGraphFromReactFlow(reloadRequest)
expect(manager.getNodes()).toEqual([
expect.objectContaining({
id: authoritativeNode.id,
data: expect.objectContaining({ title: 'Latest' }),
}),
])
expect(manager.canPersistLocalGraph()).toBe(true)
expect(sendGraphEventSpy).toHaveBeenCalledTimes(1)
expect(internals.pendingGraphResyncBroadcast).toBe(false)
unsubscribe()
manager.disconnect(connectionId)
})
it('keeps a fresh reconnect untrusted when promoted before the follower snapshot arrives', async () => {
const manager = new CollaborationManager()
const internals = getManagerInternals(manager)
const socket = createMockSocket('socket-reconnect-promotion')
const staleNode = createNode('node-stale', 'Stale')
const reactFlowStore: ReactFlowStore = {
getState: () => ({
getNodes: () => [staleNode],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
}
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
const connectionId = await manager.connect('app-reconnect-promotion', reactFlowStore)
socket.trigger('status', { isLeader: true })
socket.trigger('disconnect', 'transport close')
socket.trigger('connect')
socket.trigger('status', { isLeader: false })
expect(internals.awaitingSnapshotImport).toBe(true)
expect(internals.reconnectedWithFreshDoc).toBe(true)
expect(manager.getNodes()).toEqual([])
socket.trigger('status', { isLeader: true })
expect(internals.graphReloadRequired).toBe(true)
expect(manager.getNodes()).toEqual([])
expect(manager.canPersistLocalGraph()).toBe(false)
socket.trigger('status', { isLeader: true })
expect(manager.getNodes()).toEqual([])
expect(manager.canPersistLocalGraph()).toBe(false)
manager.disconnect(connectionId)
})
it('retries follower graph resync until a snapshot arrives', async () => {
vi.useFakeTimers()
try {
const manager = new CollaborationManager()
const socket = createMockSocket('socket-resync-retry')
const reactFlowStore: ReactFlowStore = {
getState: () => ({
getNodes: () => [],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
}
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
const connectionId = await manager.connect('app-resync-retry', reactFlowStore)
socket.trigger('status', { isLeader: false })
const getResyncRequestCount = () =>
socket.emit.mock.calls.filter(
(call) =>
call[0] === 'collaboration_event' &&
(call[1] as { type?: string } | undefined)?.type === 'graph_resync_request',
).length
expect(getResyncRequestCount()).toBe(1)
await vi.advanceTimersByTimeAsync(1000)
expect(getResyncRequestCount()).toBe(2)
const { manager: snapshotSource, internals: snapshotSourceInternals } = setupManagerWithDoc()
snapshotSource.setNodes([], [createNode('snapshot-node', 'Snapshot')])
socket.trigger('graph_update', snapshotSourceInternals.doc!.export({ mode: 'snapshot' }))
await vi.advanceTimersByTimeAsync(20_000)
expect(getResyncRequestCount()).toBe(2)
expect(manager.canPersistLocalGraph()).toBe(true)
manager.disconnect(connectionId)
} finally {
vi.useRealTimers()
}
})
it('lets a requested snapshot win over the HTTP fallback after follower promotion', async () => {
const manager = new CollaborationManager()
const socket = createMockSocket('socket-snapshot-race')
const reactFlowStore: ReactFlowStore = {
getState: () => ({
getNodes: () => [createNode('node-stale', 'Stale')],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
}
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
const connectionId = await manager.connect('app-snapshot-race', reactFlowStore)
socket.trigger('status', { isLeader: true })
socket.trigger('disconnect', 'transport close')
socket.trigger('connect')
socket.trigger('status', { isLeader: false })
let reloadRequest: GraphReloadRequest | undefined
const unsubscribe = manager.onGraphReloadRequired((request) => {
reloadRequest = request
})
socket.trigger('status', { isLeader: true })
expect(reloadRequest).toBeDefined()
const { manager: latestManager, internals: latestInternals } = setupManagerWithDoc()
latestManager.setNodes([], [createNode('node-latest', 'Latest')])
socket.trigger('graph_update', latestInternals.doc!.export({ mode: 'snapshot' }))
expect(manager.getNodes()).toEqual([
expect.objectContaining({
id: 'node-latest',
data: expect.objectContaining({ title: 'Latest' }),
}),
])
expect(manager.canPersistLocalGraph()).toBe(true)
expect(manager.isGraphReloadCurrent(reloadRequest!)).toBe(false)
unsubscribe()
manager.disconnect(connectionId)
})
it('invalidates an older HTTP reload token across another reconnect', async () => {
const manager = new CollaborationManager()
const socket = createMockSocket('socket-double-reconnect')
const reactFlowStore: ReactFlowStore = {
getState: () => ({
getNodes: () => [createNode('node-local')],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
}
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
const reloadRequests: GraphReloadRequest[] = []
const unsubscribe = manager.onGraphReloadRequired((request) => reloadRequests.push(request))
const connectionId = await manager.connect('app-double-reconnect', reactFlowStore)
socket.trigger('status', { isLeader: true })
socket.trigger('disconnect', 'transport close')
socket.trigger('connect')
socket.trigger('status', { isLeader: true })
const firstRequest = reloadRequests.at(-1)!
socket.trigger('disconnect', 'transport close')
socket.trigger('connect')
socket.trigger('status', { isLeader: true })
const secondRequest = reloadRequests.at(-1)!
expect(secondRequest.generation).toBeGreaterThan(firstRequest.generation)
expect(manager.isGraphReloadCurrent(firstRequest)).toBe(false)
expect(manager.isGraphReloadCurrent(secondRequest)).toBe(true)
unsubscribe()
manager.disconnect(connectionId)
})
it('drops rAF work scheduled by a CRDT document that was force-disconnected', () => {
const { internals } = setupManagerWithDoc()
let scheduledCallback: FrameRequestCallback | undefined
const rafSpy = vi
.spyOn(globalThis, 'requestAnimationFrame')
.mockImplementation((callback: FrameRequestCallback) => {
scheduledCallback = callback
return 1
})
const emitSpy = vi.spyOn(internals.eventEmitter, 'emit')
internals.scheduleGraphImportEmit()
internals.forceDisconnect()
emitSpy.mockClear()
scheduledCallback?.(0)
expect(emitSpy).not.toHaveBeenCalledWith('graphImport', expect.anything())
rafSpy.mockRestore()
})
it('covers setNodes/setEdges guards and destroy delegation', () => {
const { manager, internals } = setupManagerWithDoc()
const destroyDisconnectSpy = vi
@ -784,7 +1205,6 @@ describe('CollaborationManager socket and subscription behavior', () => {
const getSocketSpy = vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(null)
manager.emitCursorMove({ x: 1, y: 1, userId: 'u-1', timestamp: 1 })
manager.emitSyncRequest()
manager.emitWorkflowUpdate('app-1')
manager.emitNodePanelPresence('node-1', true, { userId: 'u-1', username: 'Alice' })
expect(sendSpy).not.toHaveBeenCalled()
@ -883,13 +1303,12 @@ describe('CollaborationManager socket and subscription behavior', () => {
.mockImplementationOnce(() => {
throw new Error('status-failed')
})
internals.crdtTrusted = false
socket.trigger('status', { isLeader: false })
expect(requestSyncSpy).toHaveBeenCalled()
expect(errorSpy).toHaveBeenCalled()
const resyncSpy = vi
.spyOn(internals, 'emitGraphResyncRequest')
.mockImplementation(() => undefined)
const resyncSpy = vi.spyOn(internals, 'emitGraphResyncRequest').mockReturnValue(true)
internals.pendingInitialSync = true
internals.isLeader = true
internals.requestInitialSyncIfNeeded()
@ -1000,9 +1419,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
expect(noLocalState[0]?.id).toBe('no-local')
internals.pendingInitialSync = false
const resyncSpy = vi
.spyOn(internals, 'emitGraphResyncRequest')
.mockImplementation(() => undefined)
const resyncSpy = vi.spyOn(internals, 'emitGraphResyncRequest').mockReturnValue(true)
privateInternals.requestInitialSyncIfNeeded()
expect(resyncSpy).not.toHaveBeenCalled()
@ -1169,6 +1586,7 @@ describe('CollaborationManager socket and subscription behavior', () => {
manager.onUndoRedoStateChange(undoStateSpy)
const connectionId = await manager.connect('app-undo-pop', reactFlowStore)
socket.trigger('status', { isLeader: true })
manager.setNodes([], nodes)
const nextNodes = nodes.map((node) => {
if (node.id === 'undo-node-1') {
@ -1206,4 +1624,142 @@ describe('CollaborationManager socket and subscription behavior', () => {
vi.useRealTimers()
rafSpy.mockRestore()
})
describe('graph view state reporting', () => {
const stubVisibilityState = (state: DocumentVisibilityState) => {
Object.defineProperty(document, 'visibilityState', {
configurable: true,
get: () => state,
})
}
afterEach(() => {
delete (document as { visibilityState?: DocumentVisibilityState }).visibilityState
})
it('emitGraphViewState sends the event and records the state when connected', () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-view-state')
internals.currentAppId = 'app-1'
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
manager.emitGraphViewState(false)
expect(internals.graphViewActive).toBe(false)
expect(socket.emit).toHaveBeenCalledTimes(1)
const [event, payload] = socket.emit.mock.calls[0] as [
string,
{ type: string; data: Record<string, unknown> },
]
expect(event).toBe('collaboration_event')
expect(payload.type).toBe('graph_view_state')
expect(payload.data).toMatchObject({ graphActive: false })
})
it('emitGraphViewState records the state without emitting when disconnected', () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-view-state-offline')
internals.currentAppId = 'app-1'
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(false)
const getSocketSpy = vi
.spyOn(webSocketClient, 'getSocket')
.mockReturnValue(socket as unknown as Socket)
manager.emitGraphViewState(false)
expect(internals.graphViewActive).toBe(false)
expect(socket.emit).not.toHaveBeenCalled()
expect(getSocketSpy).not.toHaveBeenCalled()
})
it('reports visibility changes while connected and stops after force disconnect', async () => {
const { manager, internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-visibility')
vi.spyOn(webSocketClient, 'connect').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'disconnect').mockImplementation(() => undefined)
// Invoke the captured handler directly instead of dispatching on the shared jsdom
// document, which would also trigger listeners leaked by other tests' managers.
const addListenerSpy = vi.spyOn(document, 'addEventListener')
const removeListenerSpy = vi.spyOn(document, 'removeEventListener')
stubVisibilityState('visible')
const connectionId = await manager.connect('app-visibility', {
getState: () => ({
getNodes: () => [],
setNodes: vi.fn(),
getEdges: () => [],
setEdges: vi.fn(),
}),
})
expect(internals.visibilityListenerAttached).toBe(true)
expect(internals.graphViewActive).toBe(true)
const visibilityCall = addListenerSpy.mock.calls.find(
(call) => call[0] === 'visibilitychange',
)
expect(visibilityCall).toBeDefined()
const visibilityHandler = visibilityCall?.[1] as () => void
stubVisibilityState('hidden')
visibilityHandler()
expect(internals.graphViewActive).toBe(false)
const viewStateCalls = socket.emit.mock.calls.filter(
(call) => (call[1] as { type?: string } | undefined)?.type === 'graph_view_state',
)
expect(viewStateCalls).toHaveLength(1)
expect((viewStateCalls[0]?.[1] as { data: Record<string, unknown> }).data).toMatchObject({
graphActive: false,
})
manager.disconnect(connectionId)
expect(internals.visibilityListenerAttached).toBe(false)
expect(internals.graphViewActive).toBeNull()
expect(
removeListenerSpy.mock.calls.some(
(call) => call[0] === 'visibilitychange' && call[1] === visibilityHandler,
),
).toBe(true)
})
it('re-reports both hidden and visible state after status with increasing sequences', () => {
const { internals } = setupManagerWithDoc()
const socket = createMockSocket('socket-status-rereport')
internals.currentAppId = 'app-1'
vi.spyOn(webSocketClient, 'isConnected').mockReturnValue(true)
vi.spyOn(webSocketClient, 'getSocket').mockReturnValue(socket as unknown as Socket)
internals.setupSocketEventListeners(socket as unknown as Socket)
internals.graphViewActive = false
socket.trigger('status', { isLeader: false })
const hiddenCalls = socket.emit.mock.calls.filter(
(call) => (call[1] as { type?: string } | undefined)?.type === 'graph_view_state',
)
expect(hiddenCalls).toHaveLength(1)
expect((hiddenCalls[0]?.[1] as { data: Record<string, unknown> }).data).toMatchObject({
graphActive: false,
sequence: 1,
})
socket.emit.mockClear()
internals.graphViewActive = true
socket.trigger('status', { isLeader: false })
const visibleCalls = socket.emit.mock.calls.filter(
(call) => (call[1] as { type?: string } | undefined)?.type === 'graph_view_state',
)
expect(visibleCalls).toHaveLength(1)
expect((visibleCalls[0]?.[1] as { data: Record<string, unknown> }).data).toMatchObject({
graphActive: true,
sequence: 2,
})
})
})
})

View File

@ -100,6 +100,7 @@ type CollaborationManagerInternals = {
forceDisconnect: () => void
activeConnections: Set<string>
isUndoRedoInProgress: boolean
crdtTrusted: boolean
}
const createVariable = (
@ -251,6 +252,7 @@ const setupManager = (): {
internals.doc = doc
internals.nodesMap = doc.getMap('nodes')
internals.edgesMap = doc.getMap('edges')
internals.crdtTrusted = true
return { manager, internals }
}
@ -653,6 +655,7 @@ describe('CollaborationManager public API wrappers', () => {
beforeEach(() => {
manager = new CollaborationManager()
internals = getManagerInternals(manager)
internals.crdtTrusted = true
})
it('setNodes delegates to syncNodes and commits the CRDT document', () => {

View File

@ -1,5 +1,5 @@
import type { LoroDoc } from 'loro-crdt/base64'
import type { Socket } from 'socket.io-client'
import { LoroDoc } from 'loro-crdt'
import { CRDTProvider } from '../crdt-provider'
type FakeDocEvent = {
@ -20,6 +20,7 @@ const createFakeDoc = (): FakeDoc => {
const importFn = vi.fn()
const subscribeFn = vi.fn((cb: (payload: FakeDocEvent) => void) => {
handler = cb
return vi.fn()
})
return {
@ -33,16 +34,18 @@ const createFakeDoc = (): FakeDoc => {
}
type MockSocket = {
connected: boolean
trigger: (event: string, ...args: unknown[]) => void
emit: ReturnType<typeof vi.fn>
on: ReturnType<typeof vi.fn>
off: ReturnType<typeof vi.fn>
}
const createMockSocket = (): MockSocket => {
const createMockSocket = (connected = true): MockSocket => {
const handlers = new Map<string, (...args: unknown[]) => void>()
const socket: MockSocket = {
connected,
emit: vi.fn(),
on: vi.fn((event: string, handler: (...args: unknown[]) => void) => {
handlers.set(event, handler)
@ -110,7 +113,64 @@ describe('CRDTProvider', () => {
const provider = new CRDTProvider(socket as unknown as Socket, doc as unknown as LoroDoc)
provider.destroy()
expect(socket.off).toHaveBeenCalledWith('graph_update')
expect(socket.off).toHaveBeenCalledWith('graph_update', expect.any(Function))
})
it('does not buffer local graph updates while the socket is offline', () => {
const doc = createFakeDoc()
const socket = createMockSocket(false)
const provider = new CRDTProvider(socket as unknown as Socket, doc as unknown as LoroDoc)
doc.trigger({ by: 'local' })
expect(doc.export).not.toHaveBeenCalled()
expect(socket.emit).not.toHaveBeenCalled()
provider.destroy()
})
it('reports imported snapshots after applying them', () => {
const source = new LoroDoc()
source.getMap('nodes').set('node-1', { id: 'node-1' })
source.commit()
const snapshot = source.export({ mode: 'snapshot' })
const doc = createFakeDoc()
const socket = createMockSocket()
const onSnapshotImported = vi.fn()
const provider = new CRDTProvider(
socket as unknown as Socket,
doc as unknown as LoroDoc,
undefined,
onSnapshotImported,
)
socket.trigger('graph_update', snapshot)
expect(doc.import).toHaveBeenCalledWith(snapshot)
expect(onSnapshotImported).toHaveBeenCalledTimes(1)
provider.destroy()
})
it('ignores a late snapshot when the current leader no longer accepts snapshot fallback', () => {
const source = new LoroDoc()
source.getMap('nodes').set('node-1', { id: 'node-1' })
source.commit()
const snapshot = source.export({ mode: 'snapshot' })
const doc = createFakeDoc()
const socket = createMockSocket()
const onSnapshotImported = vi.fn()
const provider = new CRDTProvider(
socket as unknown as Socket,
doc as unknown as LoroDoc,
undefined,
onSnapshotImported,
() => false,
)
socket.trigger('graph_update', snapshot)
expect(doc.import).not.toHaveBeenCalled()
expect(onSnapshotImported).not.toHaveBeenCalled()
provider.destroy()
})
it('logs an error when graph_update import fails but continues operating', () => {

View File

@ -7,11 +7,15 @@ import type {
CollaborationState,
CollaborationUpdate,
CursorPosition,
GraphReloadRequest,
NodePanelPresenceMap,
NodePanelPresenceUser,
OnlineUser,
RestoreCompleteData,
RestoreIntentData,
WorkflowSyncAcknowledgement,
WorkflowSyncRequest,
WorkflowSyncResult,
} from '../types/collaboration'
import { cloneDeep } from 'es-toolkit/object'
import { isEqual } from 'es-toolkit/predicate'
@ -127,10 +131,28 @@ type GraphSyncDiagnosticEvent = {
const GRAPH_IMPORT_LOG_LIMIT = 20
const SET_NODES_ANOMALY_LOG_LIMIT = 100
const GRAPH_SYNC_DIAGNOSTIC_LOG_LIMIT = 400
const WORKFLOW_SYNC_REQUEST_TIMEOUT = 20_000
const GRAPH_RESYNC_RETRY_BASE_DELAY = 1000
const GRAPH_RESYNC_RETRY_MAX_DELAY = 10_000
const toLoroValue = (value: unknown): Value => cloneDeep(value) as Value
const toLoroRecord = (value: unknown): Record<string, Value> =>
cloneDeep(value) as Record<string, Value>
const toUint8Array = (value: unknown): Uint8Array | null => {
if (value instanceof Uint8Array) return value
if (value instanceof ArrayBuffer) return new Uint8Array(value)
if (ArrayBuffer.isView(value))
return new Uint8Array(value.buffer, value.byteOffset, value.byteLength)
if (
Array.isArray(value) &&
value.every((item) => Number.isInteger(item) && item >= 0 && item <= 255)
)
return Uint8Array.from(value)
return null
}
export class CollaborationManager {
private doc: LoroDoc | null = null
private undoManager: UndoManager | null = null
@ -148,9 +170,30 @@ export class CollaborationManager {
private activeConnections = new Set<string>()
private isUndoRedoInProgress = false
private pendingInitialSync = false
private initialSyncRetryTimer: ReturnType<typeof setTimeout> | null = null
private initialSyncRetryAttempt = 0
private rejoinInProgress = false
private pendingGraphImportEmit = false
private pendingGraphResyncBroadcast = false
private graphViewActive: boolean | null = null
private graphViewSequence = 0
private visibilityListenerAttached = false
private crdtTrusted = false
private rebuildCrdtOnNextConnect = false
private reconnectedWithFreshDoc = false
private awaitingSnapshotImport = false
private graphReloadRequired = false
private crdtGeneration = 0
private graphReloadToken = 0
private graphReloadAttempt = 0
private pendingWorkflowSyncRequests = new Map<
string,
{
resolve: (result: WorkflowSyncResult) => void
reject: (error: Error) => void
timeout: ReturnType<typeof setTimeout>
}
>()
private graphImportLogs: GraphImportLogEntry[] = []
private setNodesAnomalyLogs: SetNodesAnomalyLogEntry[] = []
private graphSyncDiagnostics: GraphSyncDiagnosticEvent[] = []
@ -457,7 +500,7 @@ export class CollaborationManager {
newNodes: Node[],
source = 'collaboration-manager:setNodes',
): void => {
if (!this.doc) return
if (!this.doc || !this.canApplyLocalGraphMutation()) return
// Don't track operations during undo/redo to prevent loops
if (this.isUndoRedoInProgress) return
@ -469,7 +512,7 @@ export class CollaborationManager {
}
setEdges = (oldEdges: Edge[], newEdges: Edge[]): void => {
if (!this.doc) return
if (!this.doc || !this.canApplyLocalGraphMutation()) return
// Don't track operations during undo/redo to prevent loops
if (this.isUndoRedoInProgress) return
@ -483,47 +526,27 @@ export class CollaborationManager {
this.disconnect()
}
async connect(appId: string, reactFlowStore?: ReactFlowStore): Promise<string> {
const connectionId = Math.random().toString(36).substring(2, 11)
this.activeConnections.add(connectionId)
if (this.currentAppId === appId && this.doc) {
// Already connected to the same app, only update store if provided and we don't have one
if (reactFlowStore && !this.reactFlowStore) this.reactFlowStore = reactFlowStore
return connectionId
}
// Only disconnect if switching to a different app
if (this.currentAppId && this.currentAppId !== appId) this.forceDisconnect()
this.currentAppId = appId
// Only set store if provided
if (reactFlowStore) this.reactFlowStore = reactFlowStore
const socket = webSocketClient.connect(appId)
// Setup event listeners BEFORE any other operations
this.setupSocketEventListeners(socket)
private initializeCrdt(socket: Socket): void {
this.provider?.destroy()
this.undoManager = null
this.doc = new LoroDoc()
this.nodesMap = this.doc.getMap('nodes') as LoroMap<Record<string, Value>>
this.edgesMap = this.doc.getMap('edges') as LoroMap<Record<string, Value>>
this.crdtGeneration += 1
this.pendingGraphImportEmit = false
this.pendingGraphResyncBroadcast = false
this.pendingImportLog = null
// Initialize UndoManager for collaborative undo/redo
this.undoManager = new UndoManager(this.doc, {
maxUndoSteps: 100,
mergeInterval: 500, // Merge operations within 500ms
excludeOriginPrefixes: [], // Don't exclude anything - let UndoManager track all local operations
mergeInterval: 500,
excludeOriginPrefixes: [],
onPush: (_isUndo, _range, _event) => {
// Store current selection state when an operation is pushed
const selectedNode = this.reactFlowStore
?.getState()
.getNodes()
.find((n: Node) => n.data?.selected)
// Emit event to update UI button states when new operation is pushed
setTimeout(() => {
this.eventEmitter.emit('undoRedoStateChange', {
canUndo: this.undoManager?.canUndo() || false,
@ -540,7 +563,6 @@ export class CollaborationManager {
}
},
onPop: (_isUndo, value, _counterRange) => {
// Restore selection state when undoing/redoing
if (
value?.value &&
typeof value.value === 'object' &&
@ -570,9 +592,63 @@ export class CollaborationManager {
},
})
this.provider = new CRDTProvider(socket, this.doc, this.handleSessionUnauthorized)
this.provider = new CRDTProvider(
socket,
this.doc,
this.handleSessionUnauthorized,
this.handleSnapshotImported,
this.shouldImportSnapshot,
)
this.setupSubscriptions()
}
private handleSnapshotImported = (): void => {
if (!this.awaitingSnapshotImport) return
const shouldBroadcastSnapshot = this.isLeader && this.pendingGraphResyncBroadcast
this.clearInitialSyncRetry()
this.refreshGraphSynchronously()
this.awaitingSnapshotImport = false
this.crdtTrusted = true
this.reconnectedWithFreshDoc = false
this.graphReloadRequired = false
this.emitGraphReadyState()
if (shouldBroadcastSnapshot) this.broadcastCurrentGraph()
}
private shouldImportSnapshot = (): boolean => {
return !this.isLeader || this.awaitingSnapshotImport
}
async connect(appId: string, reactFlowStore?: ReactFlowStore): Promise<string> {
const connectionId = Math.random().toString(36).substring(2, 11)
this.activeConnections.add(connectionId)
if (this.currentAppId === appId && this.doc) {
// Already connected to the same app, only update store if provided and we don't have one
if (reactFlowStore && !this.reactFlowStore) this.reactFlowStore = reactFlowStore
return connectionId
}
// Only disconnect if switching to a different app
if (this.currentAppId && this.currentAppId !== appId) this.forceDisconnect()
this.currentAppId = appId
// Only set store if provided
if (reactFlowStore) this.reactFlowStore = reactFlowStore
const socket = webSocketClient.connect(appId)
// Setup event listeners BEFORE any other operations
this.setupSocketEventListeners(socket)
this.attachVisibilityListener()
this.crdtTrusted = false
this.pendingInitialSync = true
this.initializeCrdt(socket)
this.emitGraphReadyState()
// Force user_connect if already connected
if (socket.connected)
@ -593,9 +669,32 @@ export class CollaborationManager {
if (this.activeConnections.size === 0) this.forceDisconnect()
}
private rejectPendingWorkflowSyncRequests(error: Error): void {
this.pendingWorkflowSyncRequests.forEach((request) => {
clearTimeout(request.timeout)
request.reject(error)
})
this.pendingWorkflowSyncRequests.clear()
}
private forceDisconnect = (): void => {
if (this.currentAppId) webSocketClient.disconnect(this.currentAppId)
this.clearInitialSyncRetry()
this.detachVisibilityListener()
this.graphViewActive = null
this.graphViewSequence = 0
this.crdtTrusted = false
this.rebuildCrdtOnNextConnect = false
this.reconnectedWithFreshDoc = false
this.awaitingSnapshotImport = false
this.graphReloadRequired = false
this.graphReloadToken += 1
this.graphReloadAttempt = 0
this.pendingGraphResyncBroadcast = false
this.emitGraphReadyState()
this.crdtGeneration += 1
this.rejectPendingWorkflowSyncRequests(new Error('Collaboration connection closed.'))
this.provider?.destroy()
this.undoManager = null
this.doc = null
@ -635,6 +734,95 @@ export class CollaborationManager {
return this.edgesMap ? (Array.from(this.edgesMap.values()) as Edge[]) : []
}
canRestoreGraphFromCrdt(): boolean {
return this.crdtTrusted && !this.graphReloadRequired && this.doc !== null
}
canPersistLocalGraph(): boolean {
return this.crdtTrusted && !this.graphReloadRequired && this.graphViewActive !== false
}
canApplyLocalGraphMutation(): boolean {
if (!this.currentAppId) return true
return this.canPersistLocalGraph() && this.getActiveSocket()?.connected === true
}
private emitGraphReadyState(): void {
this.eventEmitter.emit('graphReadyChange', this.canApplyLocalGraphMutation())
}
onGraphReadyChange(callback: (isReady: boolean) => void): () => void {
const unsubscribe = this.eventEmitter.on('graphReadyChange', callback)
callback(this.canApplyLocalGraphMutation())
return unsubscribe
}
canFlushGraphOnPageClose(): boolean {
if (!this.crdtTrusted || this.graphReloadRequired || !this.isLeader) return false
if (this.graphViewActive !== false) return true
const socketId = this.getActiveSocket()?.id
return (
typeof socketId === 'string' &&
this.onlineUsers.length === 1 &&
this.onlineUsers[0]?.sid === socketId
)
}
private getGraphReloadRequest(): GraphReloadRequest {
return {
generation: this.crdtGeneration,
token: this.graphReloadToken,
attempt: this.graphReloadAttempt,
}
}
private emitGraphReloadRequired(attempt = 0): void {
this.graphReloadAttempt = attempt
this.graphReloadToken += 1
this.eventEmitter.emit('graphReloadRequired', this.getGraphReloadRequest())
}
onGraphReloadRequired(callback: (request: GraphReloadRequest) => void): () => void {
const unsubscribe = this.eventEmitter.on('graphReloadRequired', callback)
if (this.graphReloadRequired) callback(this.getGraphReloadRequest())
return unsubscribe
}
isGraphReloadCurrent(request: GraphReloadRequest): boolean {
return (
this.isLeader &&
this.graphReloadRequired &&
!this.crdtTrusted &&
request.generation === this.crdtGeneration &&
request.token === this.graphReloadToken
)
}
replaceGraphFromReactFlow(request: GraphReloadRequest): boolean {
if (!this.doc || !this.reactFlowStore || !this.isGraphReloadCurrent(request)) return false
const shouldBroadcastSnapshot = this.pendingGraphResyncBroadcast
const state = this.reactFlowStore.getState()
this.syncNodes(this.getNodes(), state.getNodes())
this.syncEdges(this.getEdges(), state.getEdges())
this.doc.commit()
this.clearInitialSyncRetry()
this.crdtTrusted = true
this.awaitingSnapshotImport = false
this.reconnectedWithFreshDoc = false
this.graphReloadRequired = false
this.emitGraphReadyState()
if (shouldBroadcastSnapshot) this.broadcastCurrentGraph()
return true
}
retryGraphReload(request: GraphReloadRequest): void {
if (!this.isGraphReloadCurrent(request)) return
this.emitGraphReloadRequired(request.attempt + 1)
}
emitCursorMove(position: CursorPosition): void {
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
@ -649,16 +837,108 @@ export class CollaborationManager {
})
}
emitSyncRequest(): void {
requestWorkflowSync(): Promise<WorkflowSyncResult> {
const socket = this.getActiveSocket()
if (!this.currentAppId || !socket?.connected)
return Promise.reject(new Error('Collaboration connection is not available.'))
if (!this.doc || !this.canPersistLocalGraph())
return Promise.reject(new Error('Collaborative graph is not ready to save.'))
const requestId =
globalThis.crypto?.randomUUID?.() ??
`${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`
const graphSnapshot = this.doc.export({ mode: 'snapshot' })
return new Promise<WorkflowSyncResult>((resolve, reject) => {
const timeout = setTimeout(() => {
this.pendingWorkflowSyncRequests.delete(requestId)
reject(new Error('Workflow draft sync timed out.'))
}, WORKFLOW_SYNC_REQUEST_TIMEOUT)
this.pendingWorkflowSyncRequests.set(requestId, { resolve, reject, timeout })
emitWithAuthGuard(
socket,
'collaboration_event',
{
type: 'sync_request',
data: { requestId, graphSnapshot },
timestamp: Date.now(),
} satisfies CollaborationEventPayload,
{
onAck: (body: unknown, status: unknown) => {
const pendingRequest = this.pendingWorkflowSyncRequests.get(requestId)
if (!pendingRequest) return
clearTimeout(pendingRequest.timeout)
this.pendingWorkflowSyncRequests.delete(requestId)
const response = body as {
success?: unknown
hash?: unknown
updatedAt?: unknown
error?: unknown
msg?: unknown
}
const failedStatus = typeof status === 'number' && (status < 200 || status >= 300)
if (
failedStatus ||
response?.success !== true ||
typeof response.hash !== 'string' ||
typeof response.updatedAt !== 'number'
) {
const message =
(typeof response?.error === 'string' && response.error) ||
(typeof response?.msg === 'string' && response.msg) ||
'Workflow draft sync failed.'
pendingRequest.reject(new Error(message))
return
}
pendingRequest.resolve({ hash: response.hash, updatedAt: response.updatedAt })
},
onUnauthorized: this.handleSessionUnauthorized,
},
)
})
}
emitGraphViewState(active: boolean, force = false): void {
// Record the state even while offline so the post-rejoin 'status' handler can re-report it.
if (!force && this.graphViewActive === active) return
this.graphViewActive = active
this.graphViewSequence += 1
this.emitGraphReadyState()
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
this.sendCollaborationEvent({
type: 'sync_request',
data: { timestamp: Date.now() },
type: 'graph_view_state',
data: { graphActive: active, sequence: this.graphViewSequence, timestamp: Date.now() },
timestamp: Date.now(),
})
}
private handleVisibilityChange = (): void => {
this.emitGraphViewState(document.visibilityState === 'visible')
}
private attachVisibilityListener(): void {
if (this.visibilityListenerAttached || typeof document === 'undefined') return
document.addEventListener('visibilitychange', this.handleVisibilityChange)
this.visibilityListenerAttached = true
this.graphViewActive = document.visibilityState === 'visible'
}
private detachVisibilityListener(): void {
if (!this.visibilityListenerAttached || typeof document === 'undefined') return
document.removeEventListener('visibilitychange', this.handleVisibilityChange)
this.visibilityListenerAttached = false
}
emitWorkflowUpdate(appId: string): void {
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
@ -692,7 +972,7 @@ export class CollaborationManager {
this.applyNodePanelPresenceUpdate(payload)
}
onSyncRequest(callback: () => void): () => void {
onSyncRequest(callback: (request: WorkflowSyncRequest) => void): () => void {
return this.eventEmitter.on('syncRequest', callback)
}
@ -958,7 +1238,10 @@ export class CollaborationManager {
}
private setupSubscriptions(): void {
const generation = this.crdtGeneration
this.nodesMap?.subscribe((event: LoroSubscribeEvent) => {
if (generation !== this.crdtGeneration) return
const reactFlowStore = this.reactFlowStore
const eventBy = event.by ?? 'unknown'
this.recordGraphSyncDiagnostic('nodes_subscribe', 'triggered', undefined, {
@ -986,6 +1269,8 @@ export class CollaborationManager {
this.recordGraphSyncDiagnostic('nodes_subscribe', 'queued', 'raf_scheduled')
requestAnimationFrame(() => {
if (generation !== this.crdtGeneration) return
const state = reactFlowStore.getState()
const previousNodes: Node[] = state.getNodes()
const previousEdges: Edge[] = state.getEdges()
@ -1042,6 +1327,8 @@ export class CollaborationManager {
})
this.edgesMap?.subscribe((event: LoroSubscribeEvent) => {
if (generation !== this.crdtGeneration) return
const reactFlowStore = this.reactFlowStore
const eventBy = event.by ?? 'unknown'
this.recordGraphSyncDiagnostic('edges_subscribe', 'triggered', undefined, {
@ -1069,6 +1356,8 @@ export class CollaborationManager {
this.recordGraphSyncDiagnostic('edges_subscribe', 'queued', 'raf_scheduled')
requestAnimationFrame(() => {
if (generation !== this.crdtGeneration) return
// Get ReactFlow's native setters, not the collaborative ones
const state = reactFlowStore.getState()
const previousNodes = state.getNodes()
@ -1100,7 +1389,10 @@ export class CollaborationManager {
this.recordGraphSyncDiagnostic('schedule_graph_import_emit', 'queued')
this.pendingGraphImportEmit = true
const generation = this.crdtGeneration
requestAnimationFrame(() => {
if (generation !== this.crdtGeneration) return
const beforeFinalizeNodes = this.getNodes().length
const beforeFinalizeEdges = this.getEdges().length
this.pendingGraphImportEmit = false
@ -1377,50 +1669,91 @@ export class CollaborationManager {
this.pendingImportLog = null
}
private setupSocketEventListeners(socket: Socket): void {
socket.on('collaboration_update', (update: CollaborationUpdate) => {
if (update.type === 'mouse_move') {
// Update cursor state for this user
const data = update.data as { x: number; y: number }
this.cursors[update.userId] = {
x: data.x,
y: data.y,
userId: update.userId,
timestamp: update.timestamp,
}
private importWorkflowSyncSnapshot(snapshotData: unknown): boolean {
if (!this.doc || !this.canPersistLocalGraph()) return false
this.eventEmitter.emit('cursors', { ...this.cursors })
} else if (update.type === 'vars_and_features_update') {
this.eventEmitter.emit('varsAndFeaturesUpdate', update)
} else if (update.type === 'app_state_update') {
this.eventEmitter.emit('appStateUpdate', update)
} else if (update.type === 'app_meta_update') {
this.eventEmitter.emit('appMetaUpdate', update)
} else if (update.type === 'app_publish_update') {
this.eventEmitter.emit('appPublishUpdate', update)
} else if (update.type === 'mcp_server_update') {
this.eventEmitter.emit('mcpServerUpdate', update)
} else if (update.type === 'workflow_update') {
this.eventEmitter.emit('workflowUpdate', update.data)
} else if (update.type === 'comments_update') {
this.eventEmitter.emit('commentsUpdate', update.data)
} else if (update.type === 'node_panel_presence') {
this.applyNodePanelPresenceUpdate(update.data as NodePanelPresenceEventData)
} else if (update.type === 'sync_request') {
// Only process if we are the leader
if (this.isLeader) this.eventEmitter.emit('syncRequest', {})
} else if (update.type === 'graph_resync_request') {
if (this.isLeader) this.broadcastCurrentGraph()
} else if (update.type === 'workflow_restore_intent') {
this.eventEmitter.emit('restoreIntent', update.data as RestoreIntentData)
} else if (update.type === 'workflow_restore_complete') {
this.eventEmitter.emit('restoreComplete', update.data as RestoreCompleteData)
} else if (update.type === 'workflow_history_action') {
const data = update.data as { action?: 'undo' | 'redo' | 'jump' } | undefined
if (data?.action)
this.eventEmitter.emit('historyAction', { action: data.action, userId: update.userId })
}
})
const snapshot = toUint8Array(snapshotData)
if (!snapshot) return false
try {
this.doc.import(snapshot)
return true
} catch (error) {
console.error('Error importing workflow sync snapshot:', error)
return false
}
}
private setupSocketEventListeners(socket: Socket): void {
socket.on(
'collaboration_update',
(update: CollaborationUpdate, socketAcknowledgement?: (result: unknown) => void) => {
if (update.type === 'mouse_move') {
// Update cursor state for this user
const data = update.data as { x: number; y: number }
this.cursors[update.userId] = {
x: data.x,
y: data.y,
userId: update.userId,
timestamp: update.timestamp,
}
this.eventEmitter.emit('cursors', { ...this.cursors })
} else if (update.type === 'vars_and_features_update') {
this.eventEmitter.emit('varsAndFeaturesUpdate', update)
} else if (update.type === 'app_state_update') {
this.eventEmitter.emit('appStateUpdate', update)
} else if (update.type === 'app_meta_update') {
this.eventEmitter.emit('appMetaUpdate', update)
} else if (update.type === 'app_publish_update') {
this.eventEmitter.emit('appPublishUpdate', update)
} else if (update.type === 'mcp_server_update') {
this.eventEmitter.emit('mcpServerUpdate', update)
} else if (update.type === 'workflow_update') {
this.eventEmitter.emit('workflowUpdate', update.data)
} else if (update.type === 'comments_update') {
this.eventEmitter.emit('commentsUpdate', update.data)
} else if (update.type === 'node_panel_presence') {
this.applyNodePanelPresenceUpdate(update.data as NodePanelPresenceEventData)
} else if (update.type === 'sync_request') {
const requestId =
typeof update.data.requestId === 'string' ? update.data.requestId : undefined
let acknowledged = false
const acknowledge = (result: WorkflowSyncAcknowledgement) => {
if (acknowledged || !socketAcknowledgement) return
acknowledged = true
socketAcknowledgement(result)
}
if (
!this.canPersistLocalGraph() ||
!this.importWorkflowSyncSnapshot(update.data.graphSnapshot)
) {
acknowledge({ success: false, error: 'Collaborative graph is not ready to save.' })
return
}
// The server sends sync_request only to its selected saver. The local leader flag can
// lag behind that routing decision, so receiving this directed event is the authority.
this.eventEmitter.emit('syncRequest', {
requestId,
acknowledge,
} satisfies WorkflowSyncRequest)
} else if (update.type === 'graph_resync_request') {
// The server sends this only to its selected snapshot source. Its routing decision is
// authoritative even if the preceding local leader status event is still in flight.
this.broadcastCurrentGraph()
} else if (update.type === 'workflow_restore_intent') {
this.eventEmitter.emit('restoreIntent', update.data as RestoreIntentData)
} else if (update.type === 'workflow_restore_complete') {
this.eventEmitter.emit('restoreComplete', update.data as RestoreCompleteData)
} else if (update.type === 'workflow_history_action') {
const data = update.data as { action?: 'undo' | 'redo' | 'jump' } | undefined
if (data?.action)
this.eventEmitter.emit('historyAction', { action: data.action, userId: update.userId })
}
},
)
socket.on('online_users', (data: { users: OnlineUser[]; leader?: string }) => {
try {
@ -1462,32 +1795,75 @@ export class CollaborationManager {
}
const wasLeader = this.isLeader
const wasAwaitingSnapshotImport = this.awaitingSnapshotImport
this.isLeader = data.isLeader
if (this.isLeader) {
if (
this.isLeader &&
(this.reconnectedWithFreshDoc || this.awaitingSnapshotImport || this.graphReloadRequired)
) {
this.clearInitialSyncRetry()
this.pendingInitialSync = false
const shouldNotifyReload = !this.graphReloadRequired || !wasLeader
this.graphReloadRequired = true
if (shouldNotifyReload) this.emitGraphReloadRequired()
} else if (this.isLeader) {
this.clearInitialSyncRetry()
this.seedCrdtGraphFromReactFlowIfNeeded()
this.crdtTrusted = true
this.pendingInitialSync = false
} else {
this.requestInitialSyncIfNeeded()
this.awaitingSnapshotImport = !this.crdtTrusted
this.pendingGraphResyncBroadcast = false
if (!this.awaitingSnapshotImport) {
this.clearInitialSyncRetry()
this.pendingInitialSync = false
} else if (!wasAwaitingSnapshotImport) {
this.clearInitialSyncRetry()
this.pendingInitialSync = true
this.requestInitialSyncIfNeeded()
}
}
if (wasLeader !== this.isLeader) this.eventEmitter.emit('leaderChange', this.isLeader)
this.emitGraphReadyState()
// The server recreates the session on every (re)join. Re-report both visible and hidden
// states with a monotonically increasing sequence so stale deliveries cannot win.
if (this.graphViewActive !== null) this.emitGraphViewState(this.graphViewActive, true)
} catch (error) {
console.error('Error processing status update:', error)
}
})
socket.on('connect', () => {
if (this.rebuildCrdtOnNextConnect) {
this.initializeCrdt(socket)
this.rebuildCrdtOnNextConnect = false
this.reconnectedWithFreshDoc = true
this.awaitingSnapshotImport = false
this.crdtTrusted = false
}
this.emitGraphReadyState()
this.eventEmitter.emit('stateChange', { isConnected: true })
this.pendingInitialSync = true
})
socket.on('disconnect', (reason) => {
this.clearInitialSyncRetry()
this.cursors = {}
this.onlineUsers = []
this.isLeader = false
this.leaderId = null
this.crdtTrusted = false
this.rebuildCrdtOnNextConnect = true
this.reconnectedWithFreshDoc = false
this.awaitingSnapshotImport = false
this.graphReloadRequired = false
this.pendingInitialSync = false
this.emitGraphReadyState()
this.rejectPendingWorkflowSyncRequests(new Error('Collaboration connection lost.'))
this.eventEmitter.emit('stateChange', { isConnected: false, disconnectReason: reason })
this.eventEmitter.emit('onlineUsers', [])
this.eventEmitter.emit('cursors', {})
@ -1506,25 +1882,57 @@ export class CollaborationManager {
// We currently only relay CRDT updates; the server doesn't persist them.
// When a follower joins mid-session, it might miss earlier broadcasts and render stale data.
// This lightweight checkpoint asks the leader to rebroadcast the latest graph snapshot once.
private clearInitialSyncRetry(): void {
if (this.initialSyncRetryTimer) clearTimeout(this.initialSyncRetryTimer)
this.initialSyncRetryTimer = null
this.initialSyncRetryAttempt = 0
}
private scheduleInitialSyncRetry(): void {
if (this.initialSyncRetryTimer) return
if (this.isLeader || this.crdtTrusted || !this.awaitingSnapshotImport) return
const retryDelay = Math.min(
GRAPH_RESYNC_RETRY_BASE_DELAY * 2 ** this.initialSyncRetryAttempt,
GRAPH_RESYNC_RETRY_MAX_DELAY,
)
this.initialSyncRetryTimer = setTimeout(() => {
this.initialSyncRetryTimer = null
if (this.isLeader || this.crdtTrusted || !this.awaitingSnapshotImport) {
this.initialSyncRetryAttempt = 0
return
}
this.initialSyncRetryAttempt += 1
this.pendingInitialSync = true
this.requestInitialSyncIfNeeded()
}, retryDelay)
}
private requestInitialSyncIfNeeded(): void {
if (!this.pendingInitialSync) return
if (this.isLeader) {
if (this.isLeader || this.crdtTrusted || !this.awaitingSnapshotImport) {
this.pendingInitialSync = false
this.clearInitialSyncRetry()
return
}
this.emitGraphResyncRequest()
if (!this.emitGraphResyncRequest()) return
this.pendingInitialSync = false
this.scheduleInitialSyncRetry()
}
private emitGraphResyncRequest(): void {
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
private emitGraphResyncRequest(): boolean {
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return false
if (this.getActiveSocket()?.connected !== true) return false
this.sendCollaborationEvent({
type: 'graph_resync_request',
data: { timestamp: Date.now() },
timestamp: Date.now(),
})
return true
}
private seedCrdtGraphFromReactFlowIfNeeded(): void {
@ -1551,16 +1959,20 @@ export class CollaborationManager {
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
if (!this.doc) return
if (!this.crdtTrusted || this.graphReloadRequired) {
this.pendingGraphResyncBroadcast = true
return
}
const socket = webSocketClient.getSocket(this.currentAppId)
if (!socket) return
try {
this.seedCrdtGraphFromReactFlowIfNeeded()
if (this.getNodes().length === 0 && this.getEdges().length === 0) return
const snapshot = this.doc.export({ mode: 'snapshot' })
this.sendGraphEvent(snapshot)
this.pendingGraphResyncBroadcast = false
} catch (error) {
console.error('Failed to broadcast graph snapshot:', error)
}

View File

@ -2,41 +2,71 @@
import type { LoroDoc } from 'loro-crdt'
import type { Socket } from 'socket.io-client'
import { decodeImportBlobMeta } from 'loro-crdt'
import { emitWithAuthGuard } from './websocket-manager'
export class CRDTProvider {
private doc: LoroDoc
private socket: Socket
private onUnauthorized?: () => void
private onSnapshotImported?: () => void
private shouldImportSnapshot?: () => boolean
private unsubscribeDoc?: () => void
constructor(socket: Socket, doc: LoroDoc, onUnauthorized?: () => void) {
constructor(
socket: Socket,
doc: LoroDoc,
onUnauthorized?: () => void,
onSnapshotImported?: () => void,
shouldImportSnapshot?: () => boolean,
) {
this.socket = socket
this.doc = doc
this.onUnauthorized = onUnauthorized
this.onSnapshotImported = onSnapshotImported
this.shouldImportSnapshot = shouldImportSnapshot
this.setupEventListeners()
}
private setupEventListeners(): void {
this.doc.subscribe((event: { by?: string }) => {
const unsubscribe = this.doc.subscribe((event: { by?: string }) => {
if (event.by === 'local') {
if (!this.socket.connected) return
const update = this.doc.export({ mode: 'update' })
emitWithAuthGuard(this.socket, 'graph_event', update, {
onUnauthorized: this.onUnauthorized,
})
}
})
if (typeof unsubscribe === 'function') this.unsubscribeDoc = unsubscribe
this.socket.on('graph_update', (updateData: Uint8Array) => {
this.socket.on('graph_update', this.handleGraphUpdate)
}
private handleGraphUpdate = (updateData: Uint8Array): void => {
try {
const data = new Uint8Array(updateData)
let isSnapshot = false
try {
const data = new Uint8Array(updateData)
this.doc.import(data)
} catch (error) {
console.error('Error importing graph update:', error)
const metadata = decodeImportBlobMeta(data, false)
isSnapshot = metadata.mode === 'snapshot' || metadata.mode === 'outdated-snapshot'
} catch {
// Import remains backward compatible with payloads whose metadata cannot be decoded.
}
})
if (isSnapshot && this.shouldImportSnapshot?.() === false) return
this.doc.import(data)
if (isSnapshot) this.onSnapshotImported?.()
} catch (error) {
console.error('Error importing graph update:', error)
}
}
destroy(): void {
this.socket.off('graph_update')
this.unsubscribeDoc?.()
this.unsubscribeDoc = undefined
this.socket.off('graph_update', this.handleGraphUpdate)
}
}

View File

@ -50,6 +50,7 @@ type CollaborationEventType =
| 'workflow_restore_intent'
| 'workflow_restore_complete'
| 'workflow_history_action'
| 'graph_view_state'
export type CollaborationUpdate = {
type: CollaborationEventType
@ -58,6 +59,26 @@ export type CollaborationUpdate = {
timestamp: number
}
export type WorkflowSyncResult = {
hash: string
updatedAt: number
}
export type WorkflowSyncAcknowledgement =
| ({ success: true } & WorkflowSyncResult)
| { success: false; error?: string }
export type WorkflowSyncRequest = {
requestId?: string
acknowledge: (result: WorkflowSyncAcknowledgement) => void
}
export type GraphReloadRequest = {
generation: number
token: number
attempt: number
}
export type RestoreIntentData = {
versionId: string
versionName?: string

View File

@ -26,6 +26,15 @@ export type SyncDraftCallback = {
onSettled?: () => void
}
export type SyncDraftResult = {
hash: string
updatedAt: number
}
export type SyncDraftOptions = {
forceLocal?: boolean
}
export type WorkflowAccessControl = {
canEdit: boolean
canRun: boolean
@ -44,9 +53,10 @@ type CommonHooksFnMap = {
doSyncWorkflowDraft: (
notRefreshWhenSyncError?: boolean,
callback?: SyncDraftCallback,
) => Promise<void>
options?: SyncDraftOptions,
) => Promise<SyncDraftResult | null | void>
syncWorkflowDraftWhenPageClose: () => void
handleRefreshWorkflowDraft: () => void
handleRefreshWorkflowDraft: (notUpdateCanvas?: boolean) => void
handleBackupDraft: () => void
handleLoadBackupDraft: () => void
handleRestoreFromPublishedWorkflow: (...args: any[]) => void
@ -104,7 +114,7 @@ export type Shape = {
} & CommonHooksFnMap
export const createHooksStore = ({
doSyncWorkflowDraft = async () => noop(),
doSyncWorkflowDraft = async () => null,
syncWorkflowDraftWhenPageClose = noop,
handleRefreshWorkflowDraft = noop,
handleBackupDraft = noop,

View File

@ -0,0 +1,110 @@
import type { Edge, Node } from '../../types'
import { act, renderHook } from '@testing-library/react'
import { useCollaborativeWorkflow } from '../use-collaborative-workflow'
const mocks = vi.hoisted(() => ({
canApplyLocalGraphMutation: vi.fn(),
collabSetNodes: vi.fn(),
collabSetEdges: vi.fn(),
getNodes: vi.fn(),
reactFlowSetNodes: vi.fn(),
reactFlowSetEdges: vi.fn(),
edges: [] as Edge[],
}))
vi.mock('reactflow', () => ({
useStoreApi: () => ({
getState: () => ({
getNodes: mocks.getNodes,
setNodes: mocks.reactFlowSetNodes,
edges: mocks.edges,
setEdges: mocks.reactFlowSetEdges,
}),
}),
}))
vi.mock('../../collaboration/core/collaboration-manager', () => ({
collaborationManager: {
canApplyLocalGraphMutation: mocks.canApplyLocalGraphMutation,
setNodes: mocks.collabSetNodes,
setEdges: mocks.collabSetEdges,
},
}))
const oldNode = {
id: 'old-node',
type: 'custom',
position: { x: 0, y: 0 },
data: { title: 'Old' },
} as Node
const newNode = {
...oldNode,
id: 'new-node',
data: { title: 'New' },
} as Node
const oldEdge = {
id: 'old-edge',
source: 'old-node',
target: 'new-node',
data: {},
} as Edge
const newEdge = {
...oldEdge,
id: 'new-edge',
} as Edge
describe('useCollaborativeWorkflow', () => {
beforeEach(() => {
vi.clearAllMocks()
mocks.getNodes.mockReturnValue([oldNode])
mocks.edges = [oldEdge]
})
it('drops user graph mutations while collaborative state is not ready', () => {
mocks.canApplyLocalGraphMutation.mockReturnValue(false)
const { result } = renderHook(() => useCollaborativeWorkflow())
act(() => {
result.current.setNodes([newNode])
result.current.setEdges([newEdge])
})
expect(mocks.collabSetNodes).not.toHaveBeenCalled()
expect(mocks.collabSetEdges).not.toHaveBeenCalled()
expect(mocks.reactFlowSetNodes).not.toHaveBeenCalled()
expect(mocks.reactFlowSetEdges).not.toHaveBeenCalled()
})
it('still applies non-broadcast graph imports while collaborative state is not ready', () => {
mocks.canApplyLocalGraphMutation.mockReturnValue(false)
const { result } = renderHook(() => useCollaborativeWorkflow())
act(() => {
result.current.setNodes([newNode], false)
result.current.setEdges([newEdge], false)
})
expect(mocks.collabSetNodes).not.toHaveBeenCalled()
expect(mocks.collabSetEdges).not.toHaveBeenCalled()
expect(mocks.reactFlowSetNodes).toHaveBeenCalledWith([newNode])
expect(mocks.reactFlowSetEdges).toHaveBeenCalledWith([newEdge])
})
it('updates both CRDT and ReactFlow when collaborative state is ready', () => {
mocks.canApplyLocalGraphMutation.mockReturnValue(true)
const { result } = renderHook(() => useCollaborativeWorkflow())
act(() => {
result.current.setNodes([newNode])
result.current.setEdges([newEdge])
})
expect(mocks.collabSetNodes).toHaveBeenCalledWith([oldNode], [newNode], expect.any(String))
expect(mocks.collabSetEdges).toHaveBeenCalledWith([oldEdge], [newEdge])
expect(mocks.reactFlowSetNodes).toHaveBeenCalledWith([newNode])
expect(mocks.reactFlowSetEdges).toHaveBeenCalledWith([newEdge])
})
})

View File

@ -43,6 +43,8 @@ export const useCollaborativeWorkflow = () => {
) => {
const { getNodes, setNodes: reactFlowSetNodes } = store.getState()
if (shouldBroadcast) {
if (!collaborationManager.canApplyLocalGraphMutation()) return
const oldNodes = getNodes()
collabSetNodes(
oldNodes.map(sanitizeNodeForBroadcast),
@ -59,6 +61,8 @@ export const useCollaborativeWorkflow = () => {
(newEdges: Edge[], shouldBroadcast: boolean = true) => {
const { edges, setEdges: reactFlowSetEdges } = store.getState()
if (shouldBroadcast) {
if (!collaborationManager.canApplyLocalGraphMutation()) return
collabSetEdges(edges.map(sanitizeEdgeForBroadcast), newEdges.map(sanitizeEdgeForBroadcast))
}

View File

@ -16,8 +16,9 @@ export const useNodesSyncDraft = () => {
(sync?: boolean, notRefreshWhenSyncError?: boolean, callback?: SyncDraftCallback) => {
if (getNodesReadOnly()) return
if (sync) doSyncWorkflowDraft(notRefreshWhenSyncError, callback)
else debouncedSyncWorkflowDraft(doSyncWorkflowDraft)
if (sync) return doSyncWorkflowDraft(notRefreshWhenSyncError, callback)
debouncedSyncWorkflowDraft(doSyncWorkflowDraft)
},
[debouncedSyncWorkflowDraft, doSyncWorkflowDraft, getNodesReadOnly],
)

View File

@ -384,22 +384,40 @@ export const Workflow: FC<WorkflowProps> = memo(
const { handleRefreshWorkflowDraft } = useWorkflowRefreshDraft()
const handleSyncWorkflowDraftWhenPageClose = useCallback(() => {
if (document.visibilityState === 'hidden') {
// Update the local guard synchronously. Waiting for the server's leader
// status would leave a window where this hidden tab saves a stale canvas.
collaborationManager.emitGraphViewState(false)
syncWorkflowDraftWhenPageClose()
return
}
if (document.visibilityState === 'visible') {
collaborationManager.emitGraphViewState(true)
const { isListening, workflowRunningData } = workflowStore.getState()
const status = workflowRunningData?.result?.status
// Avoid resetting UI state when user comes back while a run is active or listening for triggers
if (isListening || status === WorkflowRunningStatus.Running) return
setTimeout(() => handleRefreshWorkflowDraft(), 500)
// While this tab was hidden the canvas was frozen (rAF paused), but the CRDT doc kept
// receiving remote edits. Restore from the CRDT instead of the DB draft — the DB may
// hold the stale snapshot this very tab saved while hidden, and re-importing it would
// broadcast a rollback to everyone. A trusted CRDT remains authoritative when empty.
const collaborationConnected = collaborationManager.isConnected()
if (collaborationConnected && !collaborationManager.canRestoreGraphFromCrdt()) return
if (collaborationConnected) {
collaborationManager.refreshGraphSynchronously()
setTimeout(() => handleRefreshWorkflowDraft(true), 500)
} else {
setTimeout(() => handleRefreshWorkflowDraft(), 500)
}
}
}, [syncWorkflowDraftWhenPageClose, handleRefreshWorkflowDraft, workflowStore])
// Also add beforeunload handler as additional safety net for tab close
const handleBeforeUnload = useCallback(() => {
if (collaborationManager.canRestoreGraphFromCrdt())
collaborationManager.refreshGraphSynchronously()
syncWorkflowDraftWhenPageClose()
}, [syncWorkflowDraftWhenPageClose])

View File

@ -15,6 +15,7 @@ import VariableTrigger from '@/app/components/workflow/panel/env-panel/variable-
import { useStore } from '@/app/components/workflow/store'
const HIDDEN_SECRET_VALUE = '[__HIDDEN__]'
type DoSyncWorkflowDraft = ReturnType<typeof useNodesSyncDraft>['doSyncWorkflowDraft']
const formatSecret = (secret: string) => {
return secret.length > 8
@ -47,7 +48,7 @@ const useEnvPanelActions = ({
updateEnvList: (envList: EnvironmentVariable[]) => void
setEnvSecrets: (envSecrets: Record<string, string>) => void
setControlPromptEditorRerenderKey: (controlPromptEditorRerenderKey: number) => void
doSyncWorkflowDraft: () => Promise<void>
doSyncWorkflowDraft: DoSyncWorkflowDraft
}) => {
const emitVarsAndFeaturesUpdate = useCallback(async () => {
try {