Repository: Avaiga/taipy
Pull request: Remove in memory state in the submission
Resolves #660 

# Changes

## Core code
- Remove in-memory state in submission (`_is_abandoned`, `_is_completed`, `_is_canceled`, `_running_jobs`, `_blocked_jobs`, `_pending_jobs`) and save them to repository
- Added self setting Set class so job sets in submission can be self set
- Remove in-memory state for `_submission_entity` in `_orchestrator.py`

## Tests
- Refactor self setting and reloading attributes, properties tests in entities

Problem statement:
Remove a in-memory state
We have several places in core that relies on an in-memory state for entities. This is very error prone and a pain to maintain. 

Some places that I'm awared of:
- **_submission.py_**
```
self.__abandoned = False
self.__completed = False
self.__is_canceled = False
self.__running_jobs: MutableSet[str] = set()
self.__blocked_jobs: MutableSet[str] = set()
self.__pending_jobs: MutableSet[str] = set()
```

Hints from issue discussion:

Commit message: fixed typing

Patch to review:
diff --git a/MANIFEST.in b/MANIFEST.in
index fd021f2cbd..a5aafe110a 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -12,3 +12,42 @@ include taipy/config/*.pyi
 include taipy/config/*.json
 recursive-include taipy/templates *
 include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
diff --git a/taipy/core/_init.py b/taipy/core/_init.py
index b6dcb2988a..124fc3b123 100644
--- a/taipy/core/_init.py
+++ b/taipy/core/_init.py
@@ -42,11 +42,13 @@
     get_entities_by_config_id,
     get_jobs,
     get_latest_job,
+    get_latest_submission,
     get_parents,
     get_primary,
     get_primary_scenarios,
     get_scenarios,
     get_sequences,
+    get_submissions,
     get_tasks,
     is_deletable,
     is_editable,
diff --git a/taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py b/taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py
index 7bfb9ec2ba..9e6019dc89 100644
--- a/taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py
+++ b/taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py
@@ -29,7 +29,7 @@ class _JobDispatcher(threading.Thread):
 
     _STOP_FLAG = False
     _dispatched_processes: Dict = {}
-    __logger = _TaipyLogger._get_logger()
+    _logger = _TaipyLogger._get_logger()
     _nb_available_workers: int = 1
 
     def __init__(self, orchestrator: Optional[_AbstractOrchestrator]):
@@ -72,13 +72,13 @@ def _can_execute(self) -> bool:
     def _execute_job(self, job: Job):
         if job.force or self._needs_to_run(job.task):
             if job.force:
-                self.__logger.info(f"job {job.id} is forced to be executed.")
+                self._logger.info(f"job {job.id} is forced to be executed.")
             job.running()
             self._dispatch(job)
         else:
             job._unlock_edit_on_outputs()
             job.skipped()
-            self.__logger.info(f"job {job.id} is skipped.")
+            self._logger.info(f"job {job.id} is skipped.")
 
     def _execute_jobs_synchronously(self):
         while not self.orchestrator.jobs_to_run.empty():
@@ -86,7 +86,7 @@ def _execute_jobs_synchronously(self):
                 try:
                     job = self.orchestrator.jobs_to_run.get()
                 except Exception:  # In case the last job of the queue has been removed.
-                    self.__logger.warning(f"{job.id} is no longer in the list of jobs to run.")
+                    self._logger.warning(f"{job.id} is no longer in the list of jobs to run.")
             self._execute_job(job)
 
     @staticmethod
diff --git a/taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py b/taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py
index 18408a9b52..c03c9e1532 100644
--- a/taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py
+++ b/taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py
@@ -43,8 +43,8 @@ def _dispatch(self, job: Job):
         future = self._executor.submit(_TaskFunctionWrapper(job.id, job.task), config_as_string=config_as_string)
 
         self._set_dispatched_processes(job.id, future)  # type: ignore
-        future.add_done_callback(self._release_worker)
         future.add_done_callback(partial(self._update_job_status_from_future, job))
+        future.add_done_callback(self._release_worker)
 
     def _release_worker(self, _):
         self._nb_available_workers += 1
diff --git a/taipy/core/_orchestrator/_orchestrator.py b/taipy/core/_orchestrator/_orchestrator.py
index 7330469ba3..0d5f5693ce 100644
--- a/taipy/core/_orchestrator/_orchestrator.py
+++ b/taipy/core/_orchestrator/_orchestrator.py
@@ -14,7 +14,7 @@
 from multiprocessing import Lock
 from queue import Queue
 from time import sleep
-from typing import Callable, Dict, Iterable, List, Optional, Set, Union
+from typing import Callable, Iterable, List, Optional, Set, Union
 
 from taipy.config.config import Config
 from taipy.logger._taipy_logger import _TaipyLogger
@@ -39,7 +39,6 @@ class _Orchestrator(_AbstractOrchestrator):
     blocked_jobs: List = []
     lock = Lock()
     __logger = _TaipyLogger._get_logger()
-    _submission_entities: Dict[str, Submission] = {}
 
     @classmethod
     def initialize(cls):
@@ -66,9 +65,10 @@ def submit(
                 finished in asynchronous mode.
              timeout (Union[float, int]): The optional maximum number of seconds to wait for the jobs to be finished
                 before returning.
-             **properties (dict[str, any]): A keyworded variable length list of additional arguments.
+             **properties (dict[str, any]): A keyworded variable length list of user additional arguments
+                that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
         Returns:
-            The created Jobs.
+            The created `Submission^` containing the information about the submission.
         """
         submission = _SubmissionManagerFactory._build_manager()._create(
             submittable.id,  # type: ignore
@@ -76,7 +76,6 @@ def submit(
             getattr(submittable, "config_id", None),
             **properties,
         )
-        cls._submission_entities[submission.id] = submission
         jobs = []
         tasks = submittable._get_sorted_tasks()
         with cls.lock:
@@ -120,15 +119,15 @@ def submit_task(
                 in asynchronous mode.
              timeout (Union[float, int]): The optional maximum number of seconds to wait for the job
                 to be finished before returning.
-             **properties (dict[str, any]): A keyworded variable length list of additional arguments.
+             **properties (dict[str, any]): A keyworded variable length list of user additional arguments
+                that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
         Returns:
-            The created `Job^`.
+            The created `Submission^` containing the information about the submission.
         """
         submission = _SubmissionManagerFactory._build_manager()._create(
             task.id, task._ID_PREFIX, task.config_id, **properties
         )
         submit_id = submission.id
-        cls._submission_entities[submission.id] = submission
         with cls.lock:
             job = cls._lock_dn_output_and_create_job(
                 task,
@@ -147,10 +146,6 @@ def submit_task(
                 cls._wait_until_job_finished(job, timeout=timeout)
         return submission
 
-    @classmethod
-    def _update_submission_status(cls, job: Job):
-        cls._submission_entities[job.submit_id]._update_submission_status(job)
-
     @classmethod
     def _lock_dn_output_and_create_job(
         cls,
@@ -168,6 +163,10 @@ def _lock_dn_output_and_create_job(
 
         return job
 
+    @classmethod
+    def _update_submission_status(cls, job: Job):
+        _SubmissionManagerFactory._build_manager()._get(job.submit_id)._update_submission_status(job)
+
     @classmethod
     def _orchestrate_job_to_run_or_block(cls, jobs: List[Job]):
         blocked_jobs = []
@@ -194,7 +193,7 @@ def __check_if_timeout(st, to):
             return True
 
         start = datetime.now()
-        jobs = jobs if isinstance(jobs, Iterable) else [jobs]
+        jobs = list(jobs) if isinstance(jobs, Iterable) else [jobs]
         index = 0
         while __check_if_timeout(start, timeout) and index < len(jobs):
             try:
diff --git a/taipy/core/common/_self_setter_set.py b/taipy/core/common/_self_setter_set.py
new file mode 100644
index 0000000000..0f877deb40
--- /dev/null
+++ b/taipy/core/common/_self_setter_set.py
@@ -0,0 +1,58 @@
+# Copyright 2021-2024 Avaiga Private Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+from collections.abc import MutableSet
+from typing import Any, Iterable, Iterator
+
+
+class _SelfSetterSet(MutableSet):
+    def __init__(self, parent, data: Iterable):
+        self._parent = parent
+        self.data = set(data)
+
+    def __set_self(self):
+        from ... import core as tp
+
+        if hasattr(self, "_parent"):
+            tp.set(self._parent)
+
+    def __contains__(self, value: Any) -> bool:
+        return value in self.data
+
+    def __repr__(self) -> str:
+        return repr(self.data)
+
+    def __iter__(self) -> Iterator:
+        return iter(self.data)
+
+    def __len__(self) -> int:
+        return len(self.data)
+
+    def add(self, value: Any):
+        self.data.add(value)
+        self.__set_self()
+
+    def remove(self, value: Any):
+        self.data.remove(value)
+        self.__set_self()
+
+    def discard(self, value: Any):
+        self.data.discard(value)
+        self.__set_self()
+
+    def pop(self):
+        item = self.data.pop()
+        self.__set_self()
+        return item
+
+    def clear(self) -> None:
+        self.data.clear()
+        self.__set_self()
diff --git a/taipy/core/job/_job_manager.py b/taipy/core/job/_job_manager.py
index 7360f2fefa..b30ebdccdd 100644
--- a/taipy/core/job/_job_manager.py
+++ b/taipy/core/job/_job_manager.py
@@ -50,9 +50,11 @@ def _create(
             force=force,
             version=version,
         )
-        Notifier.publish(_make_event(job, EventOperation.CREATION))
         job._on_status_change(*callbacks)
         cls._set(job)
+
+        Notifier.publish(_make_event(job, EventOperation.CREATION))
+
         return job
 
     @classmethod
diff --git a/taipy/core/submission/_submission_converter.py b/taipy/core/submission/_submission_converter.py
index 7e04e95a40..85f9f69087 100644
--- a/taipy/core/submission/_submission_converter.py
+++ b/taipy/core/submission/_submission_converter.py
@@ -12,6 +12,7 @@
 from datetime import datetime
 
 from .._repository._abstract_converter import _AbstractConverter
+from ..common._self_setter_set import _SelfSetterSet
 from ..job.job import Job, JobId
 from ..submission._submission_model import _SubmissionModel
 from ..submission.submission import Submission
@@ -31,6 +32,12 @@ def _entity_to_model(cls, submission: Submission) -> _SubmissionModel:
             creation_date=submission._creation_date.isoformat(),
             submission_status=submission._submission_status,
             version=submission._version,
+            is_abandoned=submission._is_abandoned,
+            is_completed=submission._is_completed,
+            is_canceled=submission._is_canceled,
+            running_jobs=list(submission._running_jobs.data),
+            blocked_jobs=list(submission._blocked_jobs.data),
+            pending_jobs=list(submission._pending_jobs.data),
         )
 
     @classmethod
@@ -46,4 +53,13 @@ def _model_to_entity(cls, model: _SubmissionModel) -> Submission:
             submission_status=model.submission_status,
             version=model.version,
         )
+
+        submission._is_abandoned = model.is_abandoned
+        submission._is_completed = model.is_completed
+        submission._is_canceled = model.is_canceled
+
+        submission._running_jobs = _SelfSetterSet(submission, model.running_jobs)
+        submission._blocked_jobs = _SelfSetterSet(submission, model.blocked_jobs)
+        submission._pending_jobs = _SelfSetterSet(submission, model.pending_jobs)
+
         return submission
diff --git a/taipy/core/submission/_submission_model.py b/taipy/core/submission/_submission_model.py
index 9bc4475df1..b770716384 100644
--- a/taipy/core/submission/_submission_model.py
+++ b/taipy/core/submission/_submission_model.py
@@ -12,7 +12,7 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Union
 
-from sqlalchemy import JSON, Column, Enum, String, Table
+from sqlalchemy import JSON, Boolean, Column, Enum, String, Table
 
 from .._repository._base_taipy_model import _BaseModel
 from .._repository.db._sql_base_model import mapper_registry
@@ -35,6 +35,12 @@ class _SubmissionModel(_BaseModel):
         Column("creation_date", String),
         Column("submission_status", Enum(SubmissionStatus)),
         Column("version", String),
+        Column("is_completed", Boolean),
+        Column("is_abandoned", Boolean),
+        Column("is_canceled", Boolean),
+        Column("running_jobs", JSON),
+        Column("blocked_jobs", JSON),
+        Column("pending_jobs", JSON),
     )
     id: str
     entity_id: str
@@ -45,6 +51,12 @@ class _SubmissionModel(_BaseModel):
     creation_date: str
     submission_status: SubmissionStatus
     version: str
+    is_completed: bool
+    is_abandoned: bool
+    is_canceled: bool
+    running_jobs: List[str]
+    blocked_jobs: List[str]
+    pending_jobs: List[str]
 
     @staticmethod
     def from_dict(data: Dict[str, Any]):
@@ -58,6 +70,12 @@ def from_dict(data: Dict[str, Any]):
             creation_date=data["creation_date"],
             submission_status=SubmissionStatus._from_repr(data["submission_status"]),
             version=data["version"],
+            is_completed=data["is_completed"],
+            is_abandoned=data["is_abandoned"],
+            is_canceled=data["is_canceled"],
+            running_jobs=_BaseModel._deserialize_attribute(data["running_jobs"]),
+            blocked_jobs=_BaseModel._deserialize_attribute(data["blocked_jobs"]),
+            pending_jobs=_BaseModel._deserialize_attribute(data["pending_jobs"]),
         )
 
     def to_list(self):
@@ -71,4 +89,10 @@ def to_list(self):
             self.creation_date,
             repr(self.submission_status),
             self.version,
+            self.is_completed,
+            self.is_abandoned,
+            self.is_canceled,
+            _BaseModel._serialize_attribute(self.running_jobs),
+            _BaseModel._serialize_attribute(self.blocked_jobs),
+            _BaseModel._serialize_attribute(self.pending_jobs),
         ]
diff --git a/taipy/core/submission/submission.py b/taipy/core/submission/submission.py
index caa5887d98..f6b20adfe0 100644
--- a/taipy/core/submission/submission.py
+++ b/taipy/core/submission/submission.py
@@ -11,15 +11,15 @@
 
 import threading
 import uuid
-from collections.abc import MutableSet
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Set, Union
 
 from .._entity._entity import _Entity
 from .._entity._labeled import _Labeled
 from .._entity._properties import _Properties
 from .._entity._reload import _Reloader, _self_reload, _self_setter
 from .._version._version_manager_factory import _VersionManagerFactory
+from ..common._self_setter_set import _SelfSetterSet
 from ..job.job import Job, JobId, Status
 from ..notification.event import Event, EventEntityType, EventOperation, _make_event
 from .submission_id import SubmissionId
@@ -69,13 +69,13 @@ def __init__(
         properties = properties or {}
         self._properties = _Properties(self, **properties.copy())
 
-        self.__abandoned = False
-        self.__completed = False
+        self._is_abandoned = False
+        self._is_completed = False
+        self._is_canceled = False
 
-        self.__is_canceled = False
-        self.__running_jobs: MutableSet[str] = set()
-        self.__blocked_jobs: MutableSet[str] = set()
-        self.__pending_jobs: MutableSet[str] = set()
+        self._running_jobs: _SelfSetterSet = _SelfSetterSet(self, list())
+        self._blocked_jobs: _SelfSetterSet = _SelfSetterSet(self, list())
+        self._pending_jobs: _SelfSetterSet = _SelfSetterSet(self, list())
 
     @staticmethod
     def __new_id() -> str:
@@ -132,6 +132,21 @@ def jobs(self) -> List[Job]:
 
         return jobs
 
+    @property
+    def running_jobs(self) -> Set[Job]:
+        self._running_jobs = _Reloader()._reload(self._MANAGER_NAME, self)._running_jobs
+        return self._running_jobs
+
+    @property
+    def blocked_jobs(self) -> Set[Job]:
+        self._blocked_jobs = _Reloader()._reload(self._MANAGER_NAME, self)._blocked_jobs
+        return self._blocked_jobs
+
+    @property
+    def pending_jobs(self) -> Set[Job]:
+        self._pending_jobs = _Reloader()._reload(self._MANAGER_NAME, self)._pending_jobs
+        return self._pending_jobs
+
     @jobs.setter  # type: ignore
     @_self_setter(_MANAGER_NAME)
     def jobs(self, jobs: Union[List[Job], List[JobId]]):
@@ -145,7 +160,7 @@ def __eq__(self, other):
 
     @property  # type: ignore
     @_self_reload(_MANAGER_NAME)
-    def submission_status(self):
+    def submission_status(self) -> SubmissionStatus:
         return self._submission_status
 
     @submission_status.setter  # type: ignore
@@ -153,6 +168,36 @@ def submission_status(self):
     def submission_status(self, submission_status):
         self._submission_status = submission_status
 
+    @property  # type: ignore
+    @_self_reload(_MANAGER_NAME)
+    def is_abandoned(self) -> bool:
+        return self._is_abandoned
+
+    @is_abandoned.setter  # type: ignore
+    @_self_setter(_MANAGER_NAME)
+    def is_abandoned(self, val):
+        self._is_abandoned = val
+
+    @property  # type: ignore
+    @_self_reload(_MANAGER_NAME)
+    def is_completed(self) -> bool:
+        return self._is_completed
+
+    @is_completed.setter  # type: ignore
+    @_self_setter(_MANAGER_NAME)
+    def is_completed(self, val):
+        self._is_completed = val
+
+    @property  # type: ignore
+    @_self_reload(_MANAGER_NAME)
+    def is_canceled(self) -> bool:
+        return self._is_canceled
+
+    @is_canceled.setter  # type: ignore
+    @_self_setter(_MANAGER_NAME)
+    def is_canceled(self, val):
+        self._is_canceled = val
+
     def __lt__(self, other):
         return self.creation_date.timestamp() < other.creation_date.timestamp()
 
@@ -177,38 +222,38 @@ def _update_submission_status(self, job: Job):
 
         with self.lock:
             if job_status == Status.CANCELED:
-                self.__is_canceled = True
+                self.is_canceled = True  # type: ignore
             elif job_status == Status.BLOCKED:
-                self.__blocked_jobs.add(job.id)
-                self.__pending_jobs.discard(job.id)
+                self.blocked_jobs.add(job.id)
+                self.pending_jobs.discard(job.id)
             elif job_status == Status.PENDING or job_status == Status.SUBMITTED:
-                self.__pending_jobs.add(job.id)
-                self.__blocked_jobs.discard(job.id)
+                self.pending_jobs.add(job.id)
+                self.blocked_jobs.discard(job.id)
             elif job_status == Status.RUNNING:
-                self.__running_jobs.add(job.id)
-                self.__pending_jobs.discard(job.id)
+                self.running_jobs.add(job.id)
+                self.pending_jobs.discard(job.id)
             elif job_status == Status.COMPLETED or job_status == Status.SKIPPED:
-                self.__completed = True
-                self.__blocked_jobs.discard(job.id)
-                self.__pending_jobs.discard(job.id)
-                self.__running_jobs.discard(job.id)
+                self.is_completed = True  # type: ignore
+                self.blocked_jobs.discard(job.id)
+                self.pending_jobs.discard(job.id)
+                self.running_jobs.discard(job.id)
             elif job_status == Status.ABANDONED:
-                self.__abandoned = True
-                self.__running_jobs.discard(job.id)
-                self.__blocked_jobs.discard(job.id)
-                self.__pending_jobs.discard(job.id)
+                self.is_abandoned = True  # type: ignore
+                self.running_jobs.discard(job.id)
+                self.blocked_jobs.discard(job.id)
+                self.pending_jobs.discard(job.id)
 
-        if self.__is_canceled:
+        if self.is_canceled:
             self.submission_status = SubmissionStatus.CANCELED  # type: ignore
-        elif self.__abandoned:
+        elif self.is_abandoned:
             self.submission_status = SubmissionStatus.UNDEFINED  # type: ignore
-        elif self.__running_jobs:
+        elif self.running_jobs:
             self.submission_status = SubmissionStatus.RUNNING  # type: ignore
-        elif self.__pending_jobs:
+        elif self.pending_jobs:
             self.submission_status = SubmissionStatus.PENDING  # type: ignore
-        elif self.__blocked_jobs:
+        elif self.blocked_jobs:
             self.submission_status = SubmissionStatus.BLOCKED  # type: ignore
-        elif self.__completed:
+        elif self.is_completed:
             self.submission_status = SubmissionStatus.COMPLETED  # type: ignore
         else:
             self.submission_status = SubmissionStatus.UNDEFINED  # type: ignore
diff --git a/taipy/core/taipy.py b/taipy/core/taipy.py
index 2162777eac..de911d0488 100644
--- a/taipy/core/taipy.py
+++ b/taipy/core/taipy.py
@@ -241,8 +241,8 @@ def submit(
             in asynchronous mode.
         timeout (Union[float, int]): The optional maximum number of seconds to wait
             for the jobs to be finished before returning.
-        **properties (dict[str, any]): A keyworded variable length list of additional arguments
-            that will be stored within the `Submission^`.
+        **properties (dict[str, any]): A keyworded variable length list of user additional arguments
+            that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
 
     Returns:
         The created `Submission^` containing the information about the submission.
@@ -738,6 +738,15 @@ def get_jobs() -> List[Job]:
     return _JobManagerFactory._build_manager()._get_all()
 
 
+def get_submissions() -> List[Submission]:
+    """Return all the existing submissions.
+
+    Returns:
+        The list of all submissions.
+    """
+    return _SubmissionManagerFactory._build_manager()._get_all()
+
+
 def delete_job(job: Job, force: Optional[bool] = False):
     """Delete a job.
 
diff --git a/tests/core/_orchestrator/test_orchestrator__submit_task.py b/tests/core/_orchestrator/test_orchestrator__submit_task.py
index c28880cd60..637177c586 100644
--- a/tests/core/_orchestrator/test_orchestrator__submit_task.py
+++ b/tests/core/_orchestrator/test_orchestrator__submit_task.py
@@ -16,6 +16,7 @@
 
 from taipy.config import Config
 from taipy.core import taipy
+from taipy.core._orchestrator._orchestrator import _Orchestrator
 from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
 from taipy.core.submission._submission_manager_factory import _SubmissionManagerFactory
 from taipy.core.submission.submission_status import SubmissionStatus
@@ -62,11 +63,13 @@ def test_submit_task_development_mode():
     assert job.submit_entity_id == scenario.t1.id
     assert job.creation_date == submit_time
     assert job.stacktrace == []
-    assert len(job._subscribers) == 2  # submission._update_submission_status and orchestrator._on_status_change
+    assert len(job._subscribers) == 2  # _Orchestrator._update_submission_status and _Orchestrator._on_status_change
+    assert job._subscribers[0].__code__ == _Orchestrator._update_submission_status.__code__
+    assert job._subscribers[1].__code__ == _Orchestrator._on_status_change.__code__
 
     # submission is created and correct
     assert len(_SubmissionManagerFactory._build_manager()._get_all()) == 1
-    submission = _SubmissionManagerFactory._build_manager()._get(job.submit_id)
+    submission = _SubmissionManagerFactory._build_manager()._get(submission)
     assert submission.creation_date == submit_time
     assert submission.submission_status == SubmissionStatus.COMPLETED
     assert submission.jobs == [job]
@@ -100,12 +103,14 @@ def test_submit_task_development_mode_blocked_job():
     assert job.is_blocked()  # input data is not ready
     assert job.submit_entity_id == scenario.t2.id
     assert job.creation_date == submit_time
-    assert len(job._subscribers) == 2  # submission._update_submission_status and orchestrator._on_status_change
+    assert len(job._subscribers) == 2  # _Orchestrator._update_submission_status and _Orchestrator._on_status_change
+    assert job._subscribers[0].__code__ == _Orchestrator._update_submission_status.__code__
+    assert job._subscribers[1].__code__ == _Orchestrator._on_status_change.__code__
     assert job.stacktrace == []
 
     # submission is created and correct
     assert len(_SubmissionManagerFactory._build_manager()._get_all()) == 1
-    submission = _SubmissionManagerFactory._build_manager()._get(job.submit_id)
+    submission = _SubmissionManagerFactory._build_manager()._get(submission)
     assert submission.submission_status == SubmissionStatus.BLOCKED
     assert submission.creation_date == submit_time
     assert submission.jobs == [job]
@@ -144,12 +149,14 @@ def test_submit_task_standalone_mode():
     assert not job.force
     assert job.is_pending()
     assert job.submit_entity_id == sc.t1.id
-    assert len(job._subscribers) == 2  # submission._update_submission_status and orchestrator._on_status_change
+    assert len(job._subscribers) == 2  # _Orchestrator._update_submission_status and _Orchestrator._on_status_change
+    assert job._subscribers[0].__code__ == _Orchestrator._update_submission_status.__code__
+    assert job._subscribers[1].__code__ == _Orchestrator._on_status_change.__code__
     assert job.stacktrace == []
 
     # submission is created and correct
     assert len(_SubmissionManagerFactory._build_manager()._get_all()) == 1
-    submission = _SubmissionManagerFactory._build_manager()._get(job.submit_id)
+    submission = _SubmissionManagerFactory._build_manager()._get(submission)
     assert submission.creation_date == submit_time
     assert submission.submission_status == SubmissionStatus.PENDING
     assert submission.jobs == [job]
@@ -188,12 +195,14 @@ def test_submit_task_standalone_mode_blocked_job():
     assert not job.force
     assert job.is_blocked()  # input data is not ready
     assert job.stacktrace == []
-    assert len(job._subscribers) == 2  # submission._update_submission_status and orchestrator._on_status_change
+    assert len(job._subscribers) == 2  # _Orchestrator._update_submission_status and _Orchestrator._on_status_change
+    assert job._subscribers[0].__code__ == _Orchestrator._update_submission_status.__code__
+    assert job._subscribers[1].__code__ == _Orchestrator._on_status_change.__code__
     assert job.submit_entity_id == sc.t2.id
 
     # submission is created and correct
     assert len(_SubmissionManagerFactory._build_manager()._get_all()) == 1
-    submission = _SubmissionManagerFactory._build_manager()._get(job.submit_id)
+    submission = _SubmissionManagerFactory._build_manager()._get(submission)
     assert submission.creation_date == submit_time
     assert submission.submission_status == SubmissionStatus.BLOCKED
     assert submission.jobs == [job]
@@ -219,5 +228,11 @@ def test_submit_task_with_callbacks_and_force_and_wait():
         # job exists and is correct
         assert job.task == scenario.t1
         assert job.force
-        assert len(job._subscribers) == 3  # nothing, _update_submission_status, and _on_status_change
+        assert (
+            len(job._subscribers) == 3
+        )  # nothing, _Orchestrator._update_submission_status, and _Orchestrator._on_status_change
+        assert job._subscribers[0].__code__ == nothing.__code__
+        assert job._subscribers[1].__code__ == _Orchestrator._update_submission_status.__code__
+        assert job._subscribers[2].__code__ == _Orchestrator._on_status_change.__code__
+
         mck.assert_called_once_with(job, timeout=2)
diff --git a/tests/core/conftest.py b/tests/core/conftest.py
index 7c556a6d7b..61427ed655 100644
--- a/tests/core/conftest.py
+++ b/tests/core/conftest.py
@@ -56,6 +56,7 @@
 from taipy.core.sequence.sequence import Sequence
 from taipy.core.sequence.sequence_id import SequenceId
 from taipy.core.submission._submission_manager_factory import _SubmissionManagerFactory
+from taipy.core.submission.submission import Submission
 from taipy.core.task._task_manager_factory import _TaskManagerFactory
 from taipy.core.task.task import Task
 
@@ -285,6 +286,11 @@ def job(task):
     return Job(JobId("job"), task, "foo", "bar", version="random_version_number")
 
 
+@pytest.fixture(scope="function")
+def submission(task):
+    return Submission(task.id, task._ID_PREFIX, task.config_id, properties={})
+
+
 @pytest.fixture(scope="function")
 def _version():
     return _Version(id="foo", config=Config._applied_config)
diff --git a/tests/core/data/test_data_node.py b/tests/core/data/test_data_node.py
index 5bead24b63..5902d8865a 100644
--- a/tests/core/data/test_data_node.py
+++ b/tests/core/data/test_data_node.py
@@ -498,6 +498,57 @@ def test_auto_set_and_reload(self, current_datetime):
         assert dn_1.validity_period == time_period_2
         assert dn_2.validity_period == time_period_2
 
+        dn_1.last_edit_date = new_datetime
+
+        assert len(dn_1.job_ids) == 1
+        assert len(dn_2.job_ids) == 1
+
+        with dn_1 as dn:
+            assert dn.config_id == "foo"
+            assert dn.owner_id is None
+            assert dn.scope == Scope.SCENARIO
+            assert dn.last_edit_date == new_datetime
+            assert dn.name == "def"
+            assert dn.edit_in_progress
+            assert dn.validity_period == time_period_2
+            assert len(dn.job_ids) == 1
+            assert dn._is_in_context
+
+            new_datetime_2 = new_datetime + timedelta(5)
+
+            dn.scope = Scope.CYCLE
+            dn.last_edit_date = new_datetime_2
+            dn.name = "abc"
+            dn.edit_in_progress = False
+            dn.validity_period = None
+
+            assert dn.config_id == "foo"
+            assert dn.owner_id is None
+            assert dn.scope == Scope.SCENARIO
+            assert dn.last_edit_date == new_datetime
+            assert dn.name == "def"
+            assert dn.edit_in_progress
+            assert dn.validity_period == time_period_2
+            assert len(dn.job_ids) == 1
+
+        assert dn_1.config_id == "foo"
+        assert dn_1.owner_id is None
+        assert dn_1.scope == Scope.CYCLE
+        assert dn_1.last_edit_date == new_datetime_2
+        assert dn_1.name == "abc"
+        assert not dn_1.edit_in_progress
+        assert dn_1.validity_period is None
+        assert not dn_1._is_in_context
+        assert len(dn_1.job_ids) == 1
+
+    def test_auto_set_and_reload_properties(self):
+        dn_1 = InMemoryDataNode("foo", scope=Scope.GLOBAL, properties={"name": "def"})
+
+        dm = _DataManager()
+        dm._set(dn_1)
+
+        dn_2 = dm._get(dn_1)
+
         # auto set & reload on properties attribute
         assert dn_1.properties == {"name": "def"}
         assert dn_2.properties == {"name": "def"}
@@ -519,100 +570,37 @@ def test_auto_set_and_reload(self, current_datetime):
             "temp_key_1": "temp_value_1",
             "temp_key_2": "temp_value_2",
         }
-        assert dn_2.properties == {
-            "name": "def",
-            "qux": 5,
-            "temp_key_1": "temp_value_1",
-            "temp_key_2": "temp_value_2",
-        }
+        assert dn_2.properties == {"name": "def", "qux": 5, "temp_key_1": "temp_value_1", "temp_key_2": "temp_value_2"}
         dn_1.properties.pop("temp_key_1")
         assert "temp_key_1" not in dn_1.properties.keys()
         assert "temp_key_1" not in dn_1.properties.keys()
-        assert dn_1.properties == {
-            "name": "def",
-            "qux": 5,
-            "temp_key_2": "temp_value_2",
-        }
-        assert dn_2.properties == {
-            "name": "def",
-            "qux": 5,
-            "temp_key_2": "temp_value_2",
-        }
+        assert dn_1.properties == {"name": "def", "qux": 5, "temp_key_2": "temp_value_2"}
+        assert dn_2.properties == {"name": "def", "qux": 5, "temp_key_2": "temp_value_2"}
         dn_2.properties.pop("temp_key_2")
-        assert dn_1.properties == {
-            "qux": 5,
-            "name": "def",
-        }
-        assert dn_2.properties == {
-            "qux": 5,
-            "name": "def",
-        }
+        assert dn_1.properties == {"qux": 5, "name": "def"}
+        assert dn_2.properties == {"qux": 5, "name": "def"}
         assert "temp_key_2" not in dn_1.properties.keys()
         assert "temp_key_2" not in dn_2.properties.keys()
 
         dn_1.properties["temp_key_3"] = 0
-        assert dn_1.properties == {
-            "qux": 5,
-            "temp_key_3": 0,
-            "name": "def",
-        }
-        assert dn_2.properties == {
-            "qux": 5,
-            "temp_key_3": 0,
-            "name": "def",
-        }
+        assert dn_1.properties == {"qux": 5, "temp_key_3": 0, "name": "def"}
+        assert dn_2.properties == {"qux": 5, "temp_key_3": 0, "name": "def"}
         dn_1.properties.update({"temp_key_3": 1})
-        assert dn_1.properties == {
-            "qux": 5,
-            "temp_key_3": 1,
-            "name": "def",
-        }
-        assert dn_2.properties == {
-            "qux": 5,
-            "temp_key_3": 1,
-            "name": "def",
-        }
+        assert dn_1.properties == {"qux": 5, "temp_key_3": 1, "name": "def"}
+        assert dn_2.properties == {"qux": 5, "temp_key_3": 1, "name": "def"}
         dn_1.properties.update(dict())
-        assert dn_1.properties == {
-            "qux": 5,
-            "temp_key_3": 1,
-            "name": "def",
-        }
-        assert dn_2.properties == {
-            "qux": 5,
-            "temp_key_3": 1,
-            "name": "def",
-        }
+        assert dn_1.properties == {"qux": 5, "temp_key_3": 1, "name": "def"}
+        assert dn_2.properties == {"qux": 5, "temp_key_3": 1, "name": "def"}
         dn_1.properties["temp_key_4"] = 0
         dn_1.properties["temp_key_5"] = 0
 
-        dn_1.last_edit_date = new_datetime
-
-        assert len(dn_1.job_ids) == 1
-        assert len(dn_2.job_ids) == 1
-
         with dn_1 as dn:
-            assert dn.config_id == "foo"
-            assert dn.owner_id is None
-            assert dn.scope == Scope.SCENARIO
-            assert dn.last_edit_date == new_datetime
-            assert dn.name == "def"
-            assert dn.edit_in_progress
-            assert dn.validity_period == time_period_2
-            assert len(dn.job_ids) == 1
             assert dn._is_in_context
             assert dn.properties["qux"] == 5
             assert dn.properties["temp_key_3"] == 1
             assert dn.properties["temp_key_4"] == 0
             assert dn.properties["temp_key_5"] == 0
 
-            new_datetime_2 = new_datetime + timedelta(5)
-
-            dn.scope = Scope.CYCLE
-            dn.last_edit_date = new_datetime_2
-            dn.name = "abc"
-            dn.edit_in_progress = False
-            dn.validity_period = None
             dn.properties["qux"] = 9
             dn.properties.pop("temp_key_3")
             dn.properties.pop("temp_key_4")
@@ -621,28 +609,12 @@ def test_auto_set_and_reload(self, current_datetime):
             dn.properties.pop("temp_key_5")
             dn.properties.update(dict())
 
-            assert dn.config_id == "foo"
-            assert dn.owner_id is None
-            assert dn.scope == Scope.SCENARIO
-            assert dn.last_edit_date == new_datetime
-            assert dn.name == "def"
-            assert dn.edit_in_progress
-            assert dn.validity_period == time_period_2
-            assert len(dn.job_ids) == 1
             assert dn.properties["qux"] == 5
             assert dn.properties["temp_key_3"] == 1
             assert dn.properties["temp_key_4"] == 0
             assert dn.properties["temp_key_5"] == 0
 
-        assert dn_1.config_id == "foo"
-        assert dn_1.owner_id is None
-        assert dn_1.scope == Scope.CYCLE
-        assert dn_1.last_edit_date == new_datetime_2
-        assert dn_1.name == "abc"
-        assert not dn_1.edit_in_progress
-        assert dn_1.validity_period is None
         assert not dn_1._is_in_context
-        assert len(dn_1.job_ids) == 1
         assert dn_1.properties["qux"] == 9
         assert "temp_key_3" not in dn_1.properties.keys()
         assert dn_1.properties["temp_key_4"] == 1
diff --git a/tests/core/notification/test_events_published.py b/tests/core/notification/test_events_published.py
index eefa43f718..0ec053acb1 100644
--- a/tests/core/notification/test_events_published.py
+++ b/tests/core/notification/test_events_published.py
@@ -172,18 +172,19 @@ def test_events_published_for_scenario_submission():
     # 1 submission creation event
     # 1 submission update event for jobs
     # 3 submission update events (for status: PENDING, RUNNING and COMPLETED)
+    # 1 submission update event for is_completed
     scenario.submit()
     snapshot = all_evts.capture()
-    assert len(snapshot.collected_events) == 17
+    assert len(snapshot.collected_events) == 18
     assert snapshot.entity_type_collected.get(EventEntityType.CYCLE, 0) == 0
     assert snapshot.entity_type_collected.get(EventEntityType.DATA_NODE, 0) == 7
     assert snapshot.entity_type_collected.get(EventEntityType.TASK, 0) == 0
     assert snapshot.entity_type_collected.get(EventEntityType.SEQUENCE, 0) == 0
     assert snapshot.entity_type_collected.get(EventEntityType.SCENARIO, 0) == 1
     assert snapshot.entity_type_collected.get(EventEntityType.JOB, 0) == 4
-    assert snapshot.entity_type_collected.get(EventEntityType.SUBMISSION, 0) == 5
+    assert snapshot.entity_type_collected.get(EventEntityType.SUBMISSION, 0) == 6
     assert snapshot.operation_collected.get(EventOperation.CREATION, 0) == 2
-    assert snapshot.operation_collected.get(EventOperation.UPDATE, 0) == 14
+    assert snapshot.operation_collected.get(EventOperation.UPDATE, 0) == 15
     assert snapshot.operation_collected.get(EventOperation.SUBMISSION, 0) == 1
 
     assert snapshot.attr_name_collected["last_edit_date"] == 1
@@ -193,6 +194,7 @@ def test_events_published_for_scenario_submission():
     assert snapshot.attr_name_collected["status"] == 3
     assert snapshot.attr_name_collected["jobs"] == 1
     assert snapshot.attr_name_collected["submission_status"] == 3
+    assert snapshot.attr_name_collected["is_completed"] == 1
 
     all_evts.stop()
 
diff --git a/tests/core/scenario/test_scenario.py b/tests/core/scenario/test_scenario.py
index 7c765e64dd..f354cd789f 100644
--- a/tests/core/scenario/test_scenario.py
+++ b/tests/core/scenario/test_scenario.py
@@ -549,6 +549,75 @@ def test_auto_set_and_reload(cycle, current_datetime, task, data_node):
     assert len(scenario_1.tags) == 1
     assert len(scenario_2.tags) == 1
 
+    with scenario_1 as scenario:
+        assert scenario.config_id == "foo"
+        assert len(scenario.tasks) == 1
+        assert len(scenario.sequences) == 1
+        assert scenario.sequences["sequence_1"] == sequence_1
+        assert scenario.tasks[task.config_id] == task
+        assert len(scenario.additional_data_nodes) == 1
+        assert scenario.additional_data_nodes[additional_dn.config_id] == additional_dn
+        assert scenario.creation_date == new_datetime
+        assert scenario.cycle == cycle
+        assert scenario.is_primary
+        assert len(scenario.subscribers) == 0
+        assert len(scenario.tags) == 1
+        assert scenario._is_in_context
+        assert scenario.name == "baz"
+
+        new_datetime_2 = new_datetime + timedelta(5)
+        scenario._config_id = "foo"
+        scenario.tasks = set()
+        scenario.additional_data_nodes = set()
+        scenario.remove_sequences([sequence_1_name])
+        scenario.creation_date = new_datetime_2
+        scenario.cycle = None
+        scenario.is_primary = False
+        scenario.subscribers = [print]
+        scenario.tags = None
+        scenario.name = "qux"
+
+        assert scenario.config_id == "foo"
+        assert len(scenario.sequences) == 1
+        assert scenario.sequences[sequence_1_name] == sequence_1
+        assert len(scenario.tasks) == 1
+        assert scenario.tasks[task.config_id] == task
+        assert len(scenario.additional_data_nodes) == 1
+        assert scenario.additional_data_nodes[additional_dn.config_id] == additional_dn
+        assert scenario.creation_date == new_datetime
+        assert scenario.cycle == cycle
+        assert scenario.is_primary
+        assert len(scenario.subscribers) == 0
+        assert len(scenario.tags) == 1
+        assert scenario._is_in_context
+        assert scenario.name == "baz"
+
+    assert scenario_1.config_id == "foo"
+    assert len(scenario_1.sequences) == 0
+    assert len(scenario_1.tasks) == 0
+    assert len(scenario_1.additional_data_nodes) == 0
+    assert scenario_1.tasks == {}
+    assert scenario_1.additional_data_nodes == {}
+    assert scenario_1.creation_date == new_datetime_2
+    assert scenario_1.cycle is None
+    assert not scenario_1.is_primary
+    assert len(scenario_1.subscribers) == 1
+    assert len(scenario_1.tags) == 0
+    assert not scenario_1._is_in_context
+
+
+def test_auto_set_and_reload_properties():
+    scenario_1 = Scenario(
+        "foo",
+        set(),
+        {"name": "baz"},
+    )
+
+    scenario_manager = _ScenarioManagerFactory._build_manager()
+    scenario_manager._set(scenario_1)
+
+    scenario_2 = scenario_manager._get(scenario_1)
+
     # auto set & reload on properties attribute
     assert scenario_1.properties == {"name": "baz"}
     assert scenario_2.properties == {"name": "baz"}
@@ -608,36 +677,11 @@ def test_auto_set_and_reload(cycle, current_datetime, task, data_node):
     scenario_1.properties["temp_key_5"] = 0
 
     with scenario_1 as scenario:
-        assert scenario.config_id == "foo"
-        assert len(scenario.tasks) == 1
-        assert len(scenario.sequences) == 1
-        assert scenario.sequences["sequence_1"] == sequence_1
-        assert scenario.tasks[task.config_id] == task
-        assert len(scenario.additional_data_nodes) == 1
-        assert scenario.additional_data_nodes[additional_dn.config_id] == additional_dn
-        assert scenario.creation_date == new_datetime
-        assert scenario.cycle == cycle
-        assert scenario.is_primary
-        assert len(scenario.subscribers) == 0
-        assert len(scenario.tags) == 1
-        assert scenario._is_in_context
-        assert scenario.name == "baz"
         assert scenario.properties["qux"] == 5
         assert scenario.properties["temp_key_3"] == 1
         assert scenario.properties["temp_key_4"] == 0
         assert scenario.properties["temp_key_5"] == 0
 
-        new_datetime_2 = new_datetime + timedelta(5)
-        scenario._config_id = "foo"
-        scenario.tasks = set()
-        scenario.additional_data_nodes = set()
-        scenario.remove_sequences([sequence_1_name])
-        scenario.creation_date = new_datetime_2
-        scenario.cycle = None
-        scenario.is_primary = False
-        scenario.subscribers = [print]
-        scenario.tags = None
-        scenario.name = "qux"
         scenario.properties["qux"] = 9
         scenario.properties.pop("temp_key_3")
         scenario.properties.pop("temp_key_4")
@@ -646,36 +690,12 @@ def test_auto_set_and_reload(cycle, current_datetime, task, data_node):
         scenario.properties.pop("temp_key_5")
         scenario.properties.update(dict())
 
-        assert scenario.config_id == "foo"
-        assert len(scenario.sequences) == 1
-        assert scenario.sequences[sequence_1_name] == sequence_1
-        assert len(scenario.tasks) == 1
-        assert scenario.tasks[task.config_id] == task
-        assert len(scenario.additional_data_nodes) == 1
-        assert scenario.additional_data_nodes[additional_dn.config_id] == additional_dn
-        assert scenario.creation_date == new_datetime
-        assert scenario.cycle == cycle
-        assert scenario.is_primary
-        assert len(scenario.subscribers) == 0
-        assert len(scenario.tags) == 1
         assert scenario._is_in_context
-        assert scenario.name == "baz"
         assert scenario.properties["qux"] == 5
         assert scenario.properties["temp_key_3"] == 1
         assert scenario.properties["temp_key_4"] == 0
         assert scenario.properties["temp_key_5"] == 0
 
-    assert scenario_1.config_id == "foo"
-    assert len(scenario_1.sequences) == 0
-    assert len(scenario_1.tasks) == 0
-    assert len(scenario_1.additional_data_nodes) == 0
-    assert scenario_1.tasks == {}
-    assert scenario_1.additional_data_nodes == {}
-    assert scenario_1.creation_date == new_datetime_2
-    assert scenario_1.cycle is None
-    assert not scenario_1.is_primary
-    assert len(scenario_1.subscribers) == 1
-    assert len(scenario_1.tags) == 0
     assert not scenario_1._is_in_context
     assert scenario_1.properties["qux"] == 9
     assert "temp_key_3" not in scenario_1.properties.keys()
diff --git a/tests/core/sequence/test_sequence.py b/tests/core/sequence/test_sequence.py
index 96aff3f9dc..cd97743d93 100644
--- a/tests/core/sequence/test_sequence.py
+++ b/tests/core/sequence/test_sequence.py
@@ -582,6 +582,32 @@ def test_auto_set_and_reload(task):
     assert len(sequence_1.subscribers) == 0
     assert len(sequence_2.subscribers) == 0
 
+    with sequence_1 as sequence:
+        assert len(sequence.tasks) == 1
+        assert sequence.tasks[task.config_id].id == task.id
+        assert len(sequence.subscribers) == 0
+        assert sequence._is_in_context
+
+        sequence.tasks = []
+        sequence.subscribers = [print]
+        assert len(sequence.tasks) == 1
+        assert sequence.tasks[task.config_id].id == task.id
+        assert len(sequence.subscribers) == 0
+        assert sequence._is_in_context
+
+    assert len(sequence_1.tasks) == 0
+    assert len(sequence_1.subscribers) == 1
+    assert not sequence_1._is_in_context
+
+
+def test_auto_set_and_reload_properties():
+    scenario = Scenario("scenario", [], {}, sequences={"foo": {}})
+
+    _ScenarioManager._set(scenario)
+
+    sequence_1 = scenario.sequences["foo"]
+    sequence_2 = _SequenceManager._get(sequence_1)
+
     # auto set & reload on properties attribute
     assert sequence_1.properties == {"name": "foo"}
     assert sequence_2.properties == {"name": "foo"}
@@ -638,17 +664,12 @@ def test_auto_set_and_reload(task):
     sequence_1.properties["temp_key_5"] = 0
 
     with sequence_1 as sequence:
-        assert len(sequence.tasks) == 1
-        assert sequence.tasks[task.config_id].id == task.id
-        assert len(sequence.subscribers) == 0
         assert sequence._is_in_context
         assert sequence.properties["qux"] == 5
         assert sequence.properties["temp_key_3"] == 1
         assert sequence.properties["temp_key_4"] == 0
         assert sequence.properties["temp_key_5"] == 0
 
-        sequence.tasks = []
-        sequence.subscribers = [print]
         sequence.properties["qux"] = 9
         sequence.properties.pop("temp_key_3")
         sequence.properties.pop("temp_key_4")
@@ -657,17 +678,12 @@ def test_auto_set_and_reload(task):
         sequence.properties.pop("temp_key_5")
         sequence.properties.update(dict())
 
-        assert len(sequence.tasks) == 1
-        assert sequence.tasks[task.config_id].id == task.id
-        assert len(sequence.subscribers) == 0
         assert sequence._is_in_context
         assert sequence.properties["qux"] == 5
         assert sequence.properties["temp_key_3"] == 1
         assert sequence.properties["temp_key_4"] == 0
         assert sequence.properties["temp_key_5"] == 0
 
-    assert len(sequence_1.tasks) == 0
-    assert len(sequence_1.subscribers) == 1
     assert not sequence_1._is_in_context
     assert sequence_1.properties["qux"] == 9
     assert "temp_key_3" not in sequence_1.properties.keys()
diff --git a/tests/core/submission/test_submission.py b/tests/core/submission/test_submission.py
index ca160b09df..e832d2cd32 100644
--- a/tests/core/submission/test_submission.py
+++ b/tests/core/submission/test_submission.py
@@ -297,6 +297,36 @@ def test_auto_set_and_reload():
     assert submission_1.jobs == [job_2, job_1]
     assert submission_2.jobs == [job_2, job_1]
 
+    # auto set & reload on is_canceled attribute
+    assert not submission_1.is_canceled
+    assert not submission_2.is_canceled
+    submission_1.is_canceled = True
+    assert submission_1.is_canceled
+    assert submission_2.is_canceled
+    submission_2.is_canceled = False
+    assert not submission_1.is_canceled
+    assert not submission_2.is_canceled
+
+    # auto set & reload on is_completed attribute
+    assert not submission_1.is_completed
+    assert not submission_2.is_completed
+    submission_1.is_completed = True
+    assert submission_1.is_completed
+    assert submission_2.is_completed
+    submission_2.is_completed = False
+    assert not submission_1.is_completed
+    assert not submission_2.is_completed
+
+    # auto set & reload on is_abandoned attribute
+    assert not submission_1.is_abandoned
+    assert not submission_2.is_abandoned
+    submission_1.is_abandoned = True
+    assert submission_1.is_abandoned
+    assert submission_2.is_abandoned
+    submission_2.is_abandoned = False
+    assert not submission_1.is_abandoned
+    assert not submission_2.is_abandoned
+
     # auto set & reload on submission_status attribute
     assert submission_1.submission_status == SubmissionStatus.SUBMITTED
     assert submission_2.submission_status == SubmissionStatus.SUBMITTED
@@ -307,6 +337,97 @@ def test_auto_set_and_reload():
     assert submission_1.submission_status == SubmissionStatus.COMPLETED
     assert submission_2.submission_status == SubmissionStatus.COMPLETED
 
+    with submission_1 as submission:
+        assert submission.jobs == [job_2, job_1]
+        assert submission.submission_status == SubmissionStatus.COMPLETED
+
+        submission.jobs = [job_1]
+        submission.submission_status = SubmissionStatus.PENDING
+
+        assert submission.jobs == [job_2, job_1]
+        assert submission.submission_status == SubmissionStatus.COMPLETED
+
+    assert submission_1.jobs == [job_1]
+    assert submission_1.submission_status == SubmissionStatus.PENDING
+    assert submission_2.jobs == [job_1]
+    assert submission_2.submission_status == SubmissionStatus.PENDING
+
+
+def test_auto_set_and_reload_job_sets():
+    # pending_jobs, running_jobs, blocked_jobs have the same behavior
+    # so we will only test 1 attribute (pending_jobs) as representative
+
+    task = Task(config_id="name_1", properties={}, function=print, id=TaskId("task_1"))
+    submission_1 = Submission(task.id, task._ID_PREFIX, task.config_id, properties={})
+
+    _TaskManagerFactory._build_manager()._set(task)
+    _SubmissionManagerFactory._build_manager()._set(submission_1)
+
+    submission_2 = _SubmissionManagerFactory._build_manager()._get(submission_1)
+
+    # auto set & reload on pending_jobs attribute
+    assert len(submission_1.pending_jobs) == 0
+    assert len(submission_1.pending_jobs) == 0
+    submission_1.pending_jobs.add("job_id_1")
+    assert submission_1.pending_jobs.data == set(["job_id_1"])
+    assert submission_2.pending_jobs.data == set(["job_id_1"])
+
+    assert len(submission_1.pending_jobs) == 1
+    assert len(submission_1.pending_jobs) == 1
+    submission_2.pending_jobs.add("job_id_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2"])
+
+    submission_1.pending_jobs.add("job_id_tmp_1")
+    submission_2.pending_jobs.add("job_id_tmp_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+
+    submission_1.pending_jobs.remove("job_id_tmp_1")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_2"])
+    submission_2.pending_jobs.remove("job_id_tmp_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2"])
+
+    submission_1.pending_jobs.add("job_id_tmp_1")
+    submission_2.pending_jobs.add("job_id_tmp_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+
+    submission_1.pending_jobs.discard("job_id_tmp_1")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_2"])
+    submission_2.pending_jobs.discard("job_id_tmp_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2"])
+
+    submission_1.pending_jobs.add("job_id_tmp_1")
+    submission_2.pending_jobs.add("job_id_tmp_2")
+    assert submission_1.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+    assert submission_2.pending_jobs.data == set(["job_id_1", "job_id_2", "job_id_tmp_1", "job_id_tmp_2"])
+
+    submission_1.pending_jobs.pop()
+    assert len(submission_1.pending_jobs.data) == 3
+    assert len(submission_2.pending_jobs.data) == 3
+    submission_2.pending_jobs.pop()
+    assert len(submission_1.pending_jobs.data) == 2
+    assert len(submission_2.pending_jobs.data) == 2
+
+    submission_1.pending_jobs.clear()
+    assert len(submission_1.pending_jobs.data) == 0
+    assert len(submission_2.pending_jobs.data) == 0
+
+
+def test_auto_set_and_reload_properties():
+    task = Task(config_id="name_1", properties={}, function=print, id=TaskId("task_1"))
+    submission_1 = Submission(task.id, task._ID_PREFIX, task.config_id, properties={})
+
+    _TaskManagerFactory._build_manager()._set(task)
+    _SubmissionManagerFactory._build_manager()._set(submission_1)
+
+    submission_2 = _SubmissionManagerFactory._build_manager()._get(submission_1)
+
     # auto set & reload on properties attribute
     assert submission_1.properties == {}
     assert submission_2.properties == {}
@@ -322,27 +443,13 @@ def test_auto_set_and_reload():
 
     submission_1.properties["temp_key_1"] = "temp_value_1"
     submission_1.properties["temp_key_2"] = "temp_value_2"
-    assert submission_1.properties == {
-        "qux": 5,
-        "temp_key_1": "temp_value_1",
-        "temp_key_2": "temp_value_2",
-    }
-    assert submission_2.properties == {
-        "qux": 5,
-        "temp_key_1": "temp_value_1",
-        "temp_key_2": "temp_value_2",
-    }
+    assert submission_1.properties == {"qux": 5, "temp_key_1": "temp_value_1", "temp_key_2": "temp_value_2"}
+    assert submission_2.properties == {"qux": 5, "temp_key_1": "temp_value_1", "temp_key_2": "temp_value_2"}
     submission_1.properties.pop("temp_key_1")
     assert "temp_key_1" not in submission_1.properties.keys()
     assert "temp_key_1" not in submission_1.properties.keys()
-    assert submission_1.properties == {
-        "qux": 5,
-        "temp_key_2": "temp_value_2",
-    }
-    assert submission_2.properties == {
-        "qux": 5,
-        "temp_key_2": "temp_value_2",
-    }
+    assert submission_1.properties == {"qux": 5, "temp_key_2": "temp_value_2"}
+    assert submission_2.properties == {"qux": 5, "temp_key_2": "temp_value_2"}
     submission_2.properties.pop("temp_key_2")
     assert submission_1.properties == {"qux": 5}
     assert submission_2.properties == {"qux": 5}
@@ -362,15 +469,11 @@ def test_auto_set_and_reload():
     submission_1.properties["temp_key_5"] = 0
 
     with submission_1 as submission:
-        assert submission.jobs == [job_2, job_1]
-        assert submission.submission_status == SubmissionStatus.COMPLETED
         assert submission.properties["qux"] == 5
         assert submission.properties["temp_key_3"] == 1
         assert submission.properties["temp_key_4"] == 0
         assert submission.properties["temp_key_5"] == 0
 
-        submission.jobs = [job_1]
-        submission.submission_status = SubmissionStatus.PENDING
         submission.properties["qux"] = 9
         submission.properties.pop("temp_key_3")
         submission.properties.pop("temp_key_4")
@@ -379,17 +482,11 @@ def test_auto_set_and_reload():
         submission.properties.pop("temp_key_5")
         submission.properties.update(dict())
 
-        assert submission.jobs == [job_2, job_1]
-        assert submission.submission_status == SubmissionStatus.COMPLETED
         assert submission.properties["qux"] == 5
         assert submission.properties["temp_key_3"] == 1
         assert submission.properties["temp_key_4"] == 0
         assert submission.properties["temp_key_5"] == 0
 
-    assert submission_1.jobs == [job_1]
-    assert submission_1.submission_status == SubmissionStatus.PENDING
-    assert submission_2.jobs == [job_1]
-    assert submission_2.submission_status == SubmissionStatus.PENDING
     assert submission_1.properties["qux"] == 9
     assert "temp_key_3" not in submission_1.properties.keys()
     assert submission_1.properties["temp_key_4"] == 1
diff --git a/tests/core/task/test_task.py b/tests/core/task/test_task.py
index 64efb80014..9d77fecded 100644
--- a/tests/core/task/test_task.py
+++ b/tests/core/task/test_task.py
@@ -196,6 +196,38 @@ def test_auto_set_and_reload(data_node):
     assert task_1.parent_ids == {"sc1"}
     assert task_2.parent_ids == {"sc1"}
 
+    with task_1 as task:
+        assert task.config_id == "foo"
+        assert task.owner_id is None
+        assert task.function == mock_func
+        assert not task.skippable
+        assert task._is_in_context
+
+        task.function = print
+        task.skippable = True
+
+        assert task.config_id == "foo"
+        assert task.owner_id is None
+        assert task.function == mock_func
+        assert not task.skippable
+        assert task._is_in_context
+
+    assert task_1.config_id == "foo"
+    assert task_1.owner_id is None
+    assert task_1.function == print
+    assert task.skippable
+    assert not task_1._is_in_context
+
+
+def test_auto_set_and_reload_properties():
+    task_1 = Task(
+        config_id="foo", properties={}, function=print, input=None, output=None, owner_id=None, skippable=False
+    )
+
+    _TaskManager._set(task_1)
+
+    task_2 = _TaskManager._get(task_1)
+
     # auto set & reload on properties attribute
     assert task_1.properties == {}
     assert task_2.properties == {}
@@ -251,18 +283,12 @@ def test_auto_set_and_reload(data_node):
     task_1.properties["temp_key_5"] = 0
 
     with task_1 as task:
-        assert task.config_id == "foo"
-        assert task.owner_id is None
-        assert task.function == mock_func
-        assert not task.skippable
         assert task._is_in_context
         assert task.properties["qux"] == 5
         assert task.properties["temp_key_3"] == 1
         assert task.properties["temp_key_4"] == 0
         assert task.properties["temp_key_5"] == 0
 
-        task.function = print
-        task.skippable = True
         task.properties["qux"] = 9
         task.properties.pop("temp_key_3")
         task.properties.pop("temp_key_4")
@@ -271,20 +297,12 @@ def test_auto_set_and_reload(data_node):
         task.properties.pop("temp_key_5")
         task.properties.update(dict())
 
-        assert task.config_id == "foo"
-        assert task.owner_id is None
-        assert task.function == mock_func
-        assert not task.skippable
         assert task._is_in_context
         assert task.properties["qux"] == 5
         assert task.properties["temp_key_3"] == 1
         assert task.properties["temp_key_4"] == 0
         assert task.properties["temp_key_5"] == 0
 
-    assert task_1.config_id == "foo"
-    assert task_1.owner_id is None
-    assert task_1.function == print
-    assert task.skippable
     assert not task_1._is_in_context
     assert task_1.properties["qux"] == 9
     assert "temp_key_3" not in task_1.properties.keys()
diff --git a/tests/core/test_taipy.py b/tests/core/test_taipy.py
index 2dedb90d88..069802c6ed 100644
--- a/tests/core/test_taipy.py
+++ b/tests/core/test_taipy.py
@@ -58,7 +58,7 @@ def cb(s, j):
 
 
 class TestTaipy:
-    def test_set(self, scenario, cycle, sequence, data_node, task):
+    def test_set(self, scenario, cycle, sequence, data_node, task, submission):
         with mock.patch("taipy.core.data._data_manager._DataManager._set") as mck:
             tp.set(data_node)
             mck.assert_called_once_with(data_node)
@@ -74,6 +74,9 @@ def test_set(self, scenario, cycle, sequence, data_node, task):
         with mock.patch("taipy.core.cycle._cycle_manager._CycleManager._set") as mck:
             tp.set(cycle)
             mck.assert_called_once_with(cycle)
+        with mock.patch("taipy.core.submission._submission_manager._SubmissionManager._set") as mck:
+            tp.set(submission)
+            mck.assert_called_once_with(submission)
 
     def test_is_editable_is_called(self, cycle, job, data_node):
         with mock.patch("taipy.core.cycle._cycle_manager._CycleManager._is_editable") as mck:
@@ -569,16 +572,27 @@ def test_get_latest_job(self, task):
             tp.get_latest_job(task)
             mck.assert_called_once_with(task)
 
-    def test_get_latest_submission(self, task):
-        with mock.patch("taipy.core.submission._submission_manager._SubmissionManager._get_latest") as mck:
-            tp.get_latest_submission(task)
-            mck.assert_called_once_with(task)
-
     def test_cancel_job(self):
         with mock.patch("taipy.core.job._job_manager._JobManager._cancel") as mck:
             tp.cancel_job("job_id")
             mck.assert_called_once_with("job_id")
 
+    def test_get_submissions(self):
+        with mock.patch("taipy.core.submission._submission_manager._SubmissionManager._get_all") as mck:
+            tp.get_submissions()
+            mck.assert_called_once_with()
+
+    def test_get_submission(self, task):
+        with mock.patch("taipy.core.submission._submission_manager._SubmissionManager._get") as mck:
+            submission_id = SubmissionId("SUBMISSION_id")
+            tp.get(submission_id)
+            mck.assert_called_once_with(submission_id)
+
+    def test_get_latest_submission(self, task):
+        with mock.patch("taipy.core.submission._submission_manager._SubmissionManager._get_latest") as mck:
+            tp.get_latest_submission(task)
+            mck.assert_called_once_with(task)
+
     def test_block_config_when_core_is_running(self):
         Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE)
         input_cfg_1 = Config.configure_data_node(id="i1", storage_type="pickle", scope=Scope.SCENARIO, default_data=1)


Code review comments:
diff_hunk: @@ -12,3 +12,42 @@ include taipy/config/*.pyi
 include taipy/config/*.json
 recursive-include taipy/templates *
 include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
+include taipy/core/*.json
+include taipy/core/config/*.json
+include taipy/*.json
+include taipy/gui_core/*.json
+include taipy/gui_core/lib/*.js
+recursive-include taipy/gui/webapp *
+include taipy/gui/version.json
+include taipy/gui/viselements.json
+include taipy/gui/*.pyi
+include taipy/config/*.pyi
+include taipy/config/*.json
+recursive-include taipy/templates *
+include taipy/rest/*.json
path: MANIFEST.in
text: @user1:
Please check your configuration. You should not add these lines.

@user2:
Reading the `setup.py`, this file may change whenever we install taipy as a dependency using `pip`.