Adds client session ID propagation for user actions

Introduces client session ID to APIs and project update notifications,
enabling session-aware updates and improved multi-tab/user experience.

Allows the frontend to distinguish between updates from different
user sessions, preventing redundant UI refreshes for the originating
session when optimistic updates are used.

Relates to collaborative editing and real-time notification enhancements.

Author: matusdrobuliak66

packages/service-library/src/servicelib/rest_constants.py

@@ -23,3 +23,4 @@ class PydanticExportParametersDict(TypedDict):
 
 # Headers keys
 X_PRODUCT_NAME_HEADER: Final[str] = "X-Simcore-Products-Name"
+X_CLIENT_SESSION_ID_HEADER: Final[str] = "X-Client-Session-Id"

services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py

@@ -52,7 +52,12 @@ async def _update_project_state(
     node_errors: list[ErrorDict] | None,
 ) -> None:
     project = await _projects_service.update_project_node_state(
-        app, user_id, project_uuid, node_uuid, new_state
+        app,
+        user_id,
+        project_uuid,
+        node_uuid,
+        new_state,
+        client_session_id=None,  # <-- The trigger for this update is not from the UI (its db listener)
     )
 
     await _projects_service.notify_project_node_update(
@@ -95,6 +100,7 @@ async def _handle_db_notification(
                 changed_row.run_hash,
                 node_errors=changed_row.errors,
                 ui_changed_keys=None,
+                client_session_id=None,  # <-- The trigger for this update is not from the UI (its db listener)
             )
 
         if "state" in payload.changes and (changed_row.state is not None):

services/web/server/src/simcore_service_webserver/folders/_workspaces_repository.py

@@ -26,6 +26,7 @@ async def move_folder_into_workspace(
     folder_id: FolderID,
     workspace_id: WorkspaceID | None,
     product_name: ProductName,
+    client_session_id: str | None = None,
 ) -> None:
     # 1. User needs to have delete permission on source folder
     folder_db = await _folders_repository.get(
@@ -84,6 +85,7 @@ async def move_folder_into_workspace(
                 project_uuid=project_id,
                 patch_project_data={"workspace_id": workspace_id},
                 user_primary_gid=user["primary_gid"],
+                client_session_id=client_session_id,
             )
 
         # 5. BATCH update of folders with workspace_id

services/web/server/src/simcore_service_webserver/folders/_workspaces_rest.py

@@ -3,6 +3,7 @@
 from aiohttp import web
 from servicelib.aiohttp import status
 from servicelib.aiohttp.requests_validation import parse_request_path_parameters_as
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 
 from .._meta import api_version_prefix as VTAG
 from ..login.decorators import login_required
@@ -28,11 +29,14 @@ async def move_folder_to_workspace(request: web.Request):
     req_ctx = FoldersRequestContext.model_validate(request)
     path_params = parse_request_path_parameters_as(FolderWorkspacesPathParams, request)
 
+    client_session_id: str | None = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     await _workspaces_repository.move_folder_into_workspace(
         app=request.app,
         user_id=req_ctx.user_id,
         folder_id=path_params.folder_id,
         workspace_id=path_params.workspace_id,
         product_name=req_ctx.product_name,
+        client_session_id=client_session_id,
     )
     return web.json_response(status=status.HTTP_204_NO_CONTENT)

services/web/server/src/simcore_service_webserver/projects/_controller/nodes_rest.py

@@ -50,6 +50,7 @@
     ServiceWaitingForManualInterventionError,
     ServiceWasNotFoundError,
 )
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 from servicelib.services_utils import get_status_as_dict
 from simcore_postgres_database.models.users import UserRole
 
@@ -97,6 +98,8 @@ async def create_node(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(ProjectPathParams, request)
     body = await parse_request_body_as(NodeCreate, request)
 
+    client_session_id = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     if await _projects_service.is_service_deprecated(
         request.app,
         req_ctx.user_id,
@@ -124,6 +127,7 @@ async def create_node(request: web.Request) -> web.Response:
             body.service_key,
             body.service_version,
             body.service_id,
+            client_session_id=client_session_id,
         )
     }
     assert NodeCreated.model_validate(data) is not None  # nosec
@@ -179,6 +183,8 @@ async def patch_project_node(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(NodePathParams, request)
     node_patch = await parse_request_body_as(NodePatch, request)
 
+    client_session_id = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     await _projects_service.patch_project_node(
         request.app,
         product_name=req_ctx.product_name,
@@ -187,6 +193,7 @@ async def patch_project_node(request: web.Request) -> web.Response:
         project_id=path_params.project_id,
         node_id=path_params.node_id,
         partial_node=node_patch.to_domain_model(),
+        client_session_id=client_session_id,
     )
 
     return web.json_response(status=status.HTTP_204_NO_CONTENT)
@@ -200,6 +207,8 @@ async def delete_node(request: web.Request) -> web.Response:
     req_ctx = AuthenticatedRequestContext.model_validate(request)
     path_params = parse_request_path_parameters_as(NodePathParams, request)
 
+    client_session_id = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     # ensure the project exists
     await _projects_service.get_project_for_user(
         request.app,
@@ -213,6 +222,7 @@ async def delete_node(request: web.Request) -> web.Response:
         NodeIDStr(path_params.node_id),
         req_ctx.product_name,
         product_api_base_url=get_api_base_url(request),
+        client_session_id=client_session_id,
     )
 
     return web.json_response(status=status.HTTP_204_NO_CONTENT)
@@ -250,6 +260,8 @@ async def update_node_outputs(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(NodePathParams, request)
     node_outputs = await parse_request_body_as(NodeOutputs, request)
 
+    client_session_id = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     ui_changed_keys = set()
     ui_changed_keys.add(f"{path_params.node_id}")
     await nodes_utils.update_node_outputs(
@@ -261,6 +273,7 @@ async def update_node_outputs(request: web.Request) -> web.Response:
         run_hash=None,
         node_errors=None,
         ui_changed_keys=ui_changed_keys,
+        client_session_id=client_session_id,
     )
     return web.json_response(status=status.HTTP_204_NO_CONTENT)
 

services/web/server/src/simcore_service_webserver/projects/_controller/ports_rest.py

@@ -19,6 +19,7 @@
     parse_request_body_as,
     parse_request_path_parameters_as,
 )
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 
 from ..._meta import API_VTAG as VTAG
 from ...login.decorators import login_required
@@ -89,6 +90,8 @@ async def update_project_inputs(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(ProjectPathParams, request)
     inputs_updates = await parse_request_body_as(list[ProjectInputUpdate], request)
 
+    client_session_id: str | None = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     assert request.app  # nosec
 
     workbench = await _get_validated_workbench_model(
@@ -123,6 +126,7 @@ async def update_project_inputs(request: web.Request) -> web.Response:
         project_uuid=path_params.project_id,
         product_name=req_ctx.product_name,
         partial_workbench_data=jsonable_encoder(partial_workbench_data),
+        client_session_id=client_session_id,
     )
 
     workbench = TypeAdapter(dict[NodeID, Node]).validate_python(

services/web/server/src/simcore_service_webserver/projects/_controller/projects_rest.py

@@ -25,6 +25,7 @@
     X_SIMCORE_USER_AGENT,
 )
 from servicelib.redis import get_project_locked_state
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 
 from ..._meta import API_VTAG as VTAG
 from ...login.decorators import login_required
@@ -313,12 +314,15 @@ async def patch_project(request: web.Request):
     path_params = parse_request_path_parameters_as(ProjectPathParams, request)
     project_patch = await parse_request_body_as(ProjectPatch, request)
 
+    client_session_id: str | None = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     await _projects_service.patch_project_for_user(
         request.app,
         user_id=req_ctx.user_id,
         project_uuid=path_params.project_id,
         project_patch=project_patch,
         product_name=req_ctx.product_name,
+        client_session_id=client_session_id,
     )
 
     return web.json_response(status=status.HTTP_204_NO_CONTENT)

services/web/server/src/simcore_service_webserver/projects/_controller/workspaces_rest.py

@@ -8,6 +8,7 @@
 from pydantic import BaseModel, BeforeValidator, ConfigDict, Field
 from servicelib.aiohttp import status
 from servicelib.aiohttp.requests_validation import parse_request_path_parameters_as
+from servicelib.rest_constants import X_CLIENT_SESSION_ID_HEADER
 
 from ..._meta import api_version_prefix as VTAG
 from ...login.decorators import login_required
@@ -44,11 +45,14 @@ async def move_project_to_workspace(request: web.Request):
         _ProjectWorkspacesPathParams, request
     )
 
+    client_session_id: str | None = request.headers.get(X_CLIENT_SESSION_ID_HEADER)
+
     await _workspaces_service.move_project_into_workspace(
         app=request.app,
         user_id=req_ctx.user_id,
         project_id=path_params.project_id,
         workspace_id=path_params.workspace_id,
         product_name=req_ctx.product_name,
+        client_session_id=client_session_id,
     )
     return web.json_response(status=status.HTTP_204_NO_CONTENT)

services/web/server/src/simcore_service_webserver/projects/_projects_repository_legacy.py

@@ -881,6 +881,7 @@ async def update_project_node_data(
         node_id: NodeID,
         product_name: str | None,
         new_node_data: dict[str, Any],
+        client_session_id: str | None,
     ) -> tuple[ProjectDict, dict[NodeIDStr, Any]]:
         with log_context(
             _logger,
@@ -897,6 +898,7 @@ async def update_project_node_data(
                 project_uuid=project_uuid,
                 product_name=product_name,
                 allow_workbench_changes=False,
+                client_session_id=client_session_id,
             )
 
     async def update_project_multiple_node_data(
@@ -906,6 +908,7 @@ async def update_project_multiple_node_data(
         project_uuid: ProjectID,
         product_name: str | None,
         partial_workbench_data: dict[NodeIDStr, dict[str, Any]],
+        client_session_id: str | None,
     ) -> tuple[ProjectDict, dict[NodeIDStr, Any]]:
         """
         Raises:
@@ -923,6 +926,7 @@ async def update_project_multiple_node_data(
                 project_uuid=project_uuid,
                 product_name=product_name,
                 allow_workbench_changes=False,
+                client_session_id=client_session_id,
             )
 
     async def _update_project_workbench_with_lock_and_notify(
@@ -933,6 +937,7 @@ async def _update_project_workbench_with_lock_and_notify(
         project_uuid: ProjectID,
         product_name: str | None = None,
         allow_workbench_changes: bool,
+        client_session_id: str | None,
     ) -> tuple[ProjectDict, dict[NodeIDStr, Any]]:
         """
         Updates project workbench with Redis lock and user notification.
@@ -1002,7 +1007,7 @@ async def _update_workbench_and_notify() -> (
                 ),
             )
 
-            # Increment document version and notify users
+            # Increment document version
             redis_client_sdk = get_redis_document_manager_client_sdk(self._app)
             document_version = await increment_and_return_project_document_version(
                 redis_client=redis_client_sdk, project_uuid=project_uuid
@@ -1020,6 +1025,7 @@ async def _update_workbench_and_notify() -> (
             app=self._app,
             project_id=project_uuid,
             user_primary_gid=user_primary_gid,
+            client_session_id=client_session_id,
             version=document_version,
             document=project_document,
         )
@@ -1103,6 +1109,7 @@ async def add_project_node(
         node: ProjectNodeCreate,
         old_struct_node: Node,
         product_name: str,
+        client_session_id: str | None,
     ) -> None:
         # NOTE: permission check is done currently in update_project_workbench!
         partial_workbench_data: dict[NodeIDStr, Any] = {
@@ -1117,13 +1124,18 @@ async def add_project_node(
             project_uuid=project_id,
             product_name=product_name,
             allow_workbench_changes=True,
+            client_session_id=client_session_id,
         )
         project_nodes_repo = ProjectNodesRepo(project_uuid=project_id)
         async with self.engine.acquire() as conn:
             await project_nodes_repo.add(conn, nodes=[node])
 
     async def remove_project_node(
-        self, user_id: UserID, project_id: ProjectID, node_id: NodeID
+        self,
+        user_id: UserID,
+        project_id: ProjectID,
+        node_id: NodeID,
+        client_session_id: str | None,
     ) -> None:
         # NOTE: permission check is done currently in update_project_workbench!
         partial_workbench_data: dict[NodeIDStr, Any] = {
@@ -1134,6 +1146,7 @@ async def remove_project_node(
             user_id=user_id,
             project_uuid=project_id,
             allow_workbench_changes=True,
+            client_session_id=client_session_id,
         )
         project_nodes_repo = ProjectNodesRepo(project_uuid=project_id)
         async with self.engine.acquire() as conn: