When clearing a task do not delete draft models used by other tasks

server/services/tasks.py

@@ -815,6 +815,15 @@ def get_outputs_for_deletion(task, force=False):
             else:
                 models.draft.append(output_model)
 
+    if models.draft:
+        with TimingContext("mongo", "get_execution_models"):
+            model_ids = [m.id for m in models.draft]
+            dependent_tasks = Task.objects(execution__model__in=model_ids).only(
+                "id", "execution.model"
+            )
+            busy_models = [t.execution.model for t in dependent_tasks]
+            models.draft[:] = [m for m in models.draft if m.id not in busy_models]
+
     with TimingContext("mongo", "get_task_children"):
         tasks = Task.objects(parent=task.id).only("id", "parent", "status")
         published_tasks = [

server/tests/automated/test_tasks_edit.py

@@ -1,3 +1,4 @@
+from apierrors.errors.bad_request import InvalidModelId
 from config import config
 from tests.automated import TestService
 
@@ -38,6 +39,23 @@ class TestTasksEdit(TestService):
         self.assertFalse(self.api.models.get_by_id(model=not_ready_model).model.ready)
         self.api.tasks.edit(task=task, execution=dict(model=not_ready_model))
 
+    def test_task_with_model_reset(self):
+        # on task reset output model deleted
+        task = self.new_task()
+        self.api.tasks.started(task=task)
+        model_id = self.api.models.update_for_task(task=task, uri="file:///b")["id"]
+        self.api.tasks.reset(task=task)
+        with self.api.raises(InvalidModelId):
+            self.api.models.get_by_id(model=model_id)
+
+        # unless it is input of some task
+        task = self.new_task()
+        self.api.tasks.started(task=task)
+        model_id = self.api.models.update_for_task(task=task, uri="file:///b")["id"]
+        task_2 = self.new_task(execution=dict(model=model_id))
+        self.api.tasks.reset(task=task)
+        self.api.models.get_by_id(model=model_id)
+
     def test_clone_task(self):
         script = dict(
             binary="python",