Repository: keephq/keep
Pull request: fix: delete workflows that are not provisioned anymore
<!-- 
Thanks for creating this pull request 🤗

Please make sure that the pull request is limited to one type (docs, feature, etc.) and keep it as small as possible. You can open multiple prs instead of opening a huge one.
-->

<!-- If this pull request closes an issue, please mention the issue number below -->
Closes #4603  <!-- Issue # here -->

Problem statement:
[🐛 Bug]: Workflow not deleted when we don't provision it anymore
Hi team,

I am provisioning a workflow via the environment, but when I stop provisioning it, the workflow continues to run normally instead of being deleted.

Additionally, I’ve noticed that if I repeatedly stop provisioning and restart the backend, the number of workflows increases significantly.

Hints from issue discussion:
<!-- Answer -->
When a workflow is no longer provisioned, it should be deprovisioned based on specific conditions. The system checks if the workflow's provisioned file is `None`, does not exist, or is outside the `provisioned_workflows_dir`. If any of these conditions are met, the workflow should be deprovisioned <sup>[[1]](https://github.com/keephq/keep/blob/main/keep/workflowmanager/workflowstore.py)</sup>.

If the workflow continues to run despite not being provisioned, ensure that the `KEEP_WORKFLOWS_DIRECTORY` environment variable is correctly set and that the application is restarted to trigger the deprovisioning process <sup>[[2]](https://github.com/keephq/keep/blob/main/docs/deployment/provision/workflow.mdx)</sup>.

Regarding the increase in the number of workflows, this might be due to the system not correctly identifying and deprovisioning orphaned workflows. Ensure that the backend is properly configured to check for and remove orphaned workflows during restarts <sup>[[1]](https://github.com/keephq/keep/blob/main/keep/workflowmanager/workflowstore.py)</sup><sup>[[3]](https://github.com/keephq/keep/blob/main/keep/api/models/db/migrations/versions/2025-04-15-15-30_885ff6b12fed.py)</sup>.


<!-- Dosu Comment Footer -->
*To continue talking to [Dosu](https://dosu.dev), mention @dosu.*

---
<sup>Help Dosu learn! Give it feedback: [Great Response](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=great_response) | [Irrelevant Answer](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=irrelevant_answer) | [Incorrect Sources](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=incorrect_sources) | [Too Verbose](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=hallucination) | [Bug Report](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/16121d7f-633d-4872-8cbb-cfaaf48986ce?feedback_type=other)</sup>

Can I work on it?
Done
Hi @EnotShow, I have figured out the new bug in recent versions of Keep. Whenever the backend service restarts, the new duplicated workflow is provisioned one more time. Therefore, this will result tons of workflows.

<img width="759" alt="Image" src="https://github.com/user-attachments/assets/e9e53fa8-c423-49c7-ada8-3b60b8d894df" />

Hi @tuantran0910, thanks for information, but I've already fix it in current opened PR
> Hi [@tuantran0910](https://github.com/tuantran0910), thanks for information, but I've already fix it in current opened PR

Nice @EnotShow, some unit tests failed. You may need to have a look at them in order for the team to review and hopefully merge.
Ok, I see when I missed. I'll fix it little bit later today

Commit message: fix: use "workflow" key value as workflow_dict if present

Patch to review:
diff --git a/keep/api/core/db.py b/keep/api/core/db.py
index 5dc0f14dc6..85a3f15faa 100644
--- a/keep/api/core/db.py
+++ b/keep/api/core/db.py
@@ -481,7 +481,7 @@ def update_workflow_with_values(
     existing_workflow: Workflow,
     name: str,
     description: str | None,
-    interval: int,
+    interval: int | None,
     workflow_raw: str,
     is_disabled: bool,
     updated_by: str,
@@ -544,7 +544,7 @@ def add_or_update_workflow(
     tenant_id: str,
     description: str | None,
     created_by: str,
-    interval: int,
+    interval: int | None,
     workflow_raw: str,
     is_disabled: bool,
     updated_by: str,
diff --git a/keep/api/models/workflow.py b/keep/api/models/workflow.py
index ef618e0547..19362e1c65 100644
--- a/keep/api/models/workflow.py
+++ b/keep/api/models/workflow.py
@@ -146,3 +146,11 @@ class WorkflowVersionDTO(BaseModel):
 
 class WorkflowVersionListDTO(BaseModel):
     versions: List[WorkflowVersionDTO]
+
+
+class PreparsedWorkflowDTO(BaseModel):
+    id: str
+    name: str
+    description: Optional[str] = "Workflow file doesn't contain description"
+    interval: int | None = None
+    disabled: bool = False
diff --git a/keep/parser/parser.py b/keep/parser/parser.py
index 4d8716574c..3c3627b366 100644
--- a/keep/parser/parser.py
+++ b/keep/parser/parser.py
@@ -184,7 +184,7 @@ def _parse_workflow(
         )
         workflow_interval = self.parse_interval(workflow)
         on_failure_action = self._get_on_failure_action(context_manager, workflow)
-        workflow_triggers = self.get_triggers_from_workflow(workflow)
+        workflow_triggers = self.get_triggers_from_workflow_dict(workflow)
         workflow_provider_types = (
             self._get_workflow_provider_types_from_steps_and_actions(
                 workflow_steps, workflow_actions
@@ -833,7 +833,7 @@ def _parse_provider_config(
                 provider_config = {"authentication": {}}
             return config_id, provider_config
 
-    def get_providers_from_workflow(self, workflow: dict):
+    def get_providers_from_workflow_dict(self, workflow: dict):
         """extract the provider names from a worklow
 
         Args:
@@ -869,7 +869,7 @@ def get_providers_from_workflow(self, workflow: dict):
             raise
         return providers
 
-    def get_triggers_from_workflow(self, workflow: dict):
+    def get_triggers_from_workflow_dict(self, workflow: dict):
         """extract the trigger names from a worklow
 
         Args:
diff --git a/keep/workflowmanager/workflowstore.py b/keep/workflowmanager/workflowstore.py
index d79c954909..d69b10f308 100644
--- a/keep/workflowmanager/workflowstore.py
+++ b/keep/workflowmanager/workflowstore.py
@@ -21,7 +21,7 @@
 )
 from keep.api.core.workflows import get_workflows_with_last_executions_v2
 from keep.api.models.db.workflow import Workflow as WorkflowModel
-from keep.api.models.workflow import ProviderDTO
+from keep.api.models.workflow import PreparsedWorkflowDTO, ProviderDTO
 from keep.functions import cyaml
 from keep.parser.parser import Parser
 from keep.providers.providers_factory import ProvidersFactory
@@ -277,19 +277,19 @@ def pre_parse_workflow_yaml(yaml_content):
 
         workflow_name = yaml_content.get("name") or yaml_content.get("id")
         if not workflow_name:
-            raise Exception(f"Workflow {yaml_content} does not have a name or id")
+            raise ValueError(f"Workflow {yaml_content} does not have a name or id")
 
         workflow_id = str(uuid.uuid4())
         workflow_description = yaml_content.get("description")
         workflow_interval = parser.parse_interval(yaml_content)
         workflow_disabled = parser.parse_disabled(yaml_content)
 
-        return (
-            workflow_id,
-            workflow_name,
-            workflow_description,
-            workflow_interval,
-            workflow_disabled,
+        return PreparsedWorkflowDTO(
+            id=workflow_id,
+            name=workflow_name,
+            description=workflow_description,
+            interval=workflow_interval,
+            disabled=workflow_disabled,
         )
 
     @staticmethod
@@ -311,8 +311,18 @@ def provision_workflows(
         provisioned_workflows_dir = os.environ.get("KEEP_WORKFLOWS_DIRECTORY")
         provisioned_workflow_yaml = os.environ.get("KEEP_WORKFLOW")
 
+        # Get all existing provisioned workflows
+        provisioned_workflows = get_all_provisioned_workflows(tenant_id)
+
         if not (provisioned_workflows_dir or provisioned_workflow_yaml):
             logger.info("No workflows for provisioning found")
+
+            if provisioned_workflows:
+                logger.info("Found existing provisioned workflows, deleting them")
+                for workflow in provisioned_workflows:
+                    logger.info(f"Deprovisioning workflow {workflow.id}")
+                    delete_workflow(tenant_id, workflow.id)
+                    logger.info(f"Workflow {workflow.id} deprovisioned successfully")
             return []
 
         if (
@@ -330,43 +340,53 @@ def provision_workflows(
                 f"Directory {provisioned_workflows_dir} does not exist"
             )
 
-        # Get all existing provisioned workflows
-        provisioned_workflows = get_all_provisioned_workflows(tenant_id)
-
         ### Provisioning from env var
         if provisioned_workflow_yaml is not None:
+            pre_parsed_workflow = None
             try:
                 workflow_yaml = cyaml.safe_load(provisioned_workflow_yaml)
-                (
-                    workflow_id,
-                    workflow_name,
-                    workflow_description,
-                    workflow_interval,
-                    workflow_disabled,
-                ) = WorkflowStore.pre_parse_workflow_yaml(workflow_yaml)
+                pre_parsed_workflow = WorkflowStore.pre_parse_workflow_yaml(
+                    workflow_yaml
+                )
+            except ValueError as e:
+                logger.error(
+                    "Error provisioning workflow from env var: yaml is invalid",
+                    extra={"exception": e},
+                )
 
+            try:
                 # Un-provisioning other workflows.
                 for workflow in provisioned_workflows:
-                    if not workflow.id == workflow_id:
-                        logger.info(
-                            f"Deprovisioning workflow {workflow.id} as its id doesn't match the provisioned workflow provided in the env"
-                        )
-                        delete_workflow_by_provisioned_file(
-                            tenant_id, workflow.provisioned_file
-                        )
+                    if (
+                        not pre_parsed_workflow
+                        or not workflow.id == pre_parsed_workflow.id
+                    ):
+                        if not pre_parsed_workflow:
+                            logger.info(
+                                f"Deprovisioning workflow {workflow.id} as no workflows to provision"
+                            )
+                        else:
+                            logger.info(
+                                f"Deprovisioning workflow {workflow.id} as its id doesn't match the provisioned workflow provided in the env"
+                            )
+                        delete_workflow(tenant_id, workflow.id)
                         logger.info(
                             f"Workflow {workflow.id} deprovisioned successfully"
                         )
 
+                if not pre_parsed_workflow:
+                    logger.info("No workflows to provision")
+                    return []
+
                 add_or_update_workflow(
-                    id=workflow_id,
-                    name=workflow_name,
+                    id=pre_parsed_workflow.id,
+                    name=pre_parsed_workflow.name,
                     tenant_id=tenant_id,
-                    description=workflow_description,
+                    description=pre_parsed_workflow.description,
                     created_by="system",
                     updated_by="system",
-                    interval=workflow_interval,
-                    is_disabled=workflow_disabled,
+                    interval=pre_parsed_workflow.interval,
+                    is_disabled=pre_parsed_workflow.disabled,
                     workflow_raw=cyaml.dump(workflow_yaml, width=99999),
                     provisioned=True,
                     provisioned_file=None,
@@ -410,22 +430,18 @@ def provision_workflows(
                     try:
                         with open(workflow_path, "r") as yaml_file:
                             workflow_yaml = cyaml.safe_load(yaml_file.read())
-                            (
-                                workflow_id,
-                                workflow_name,
-                                workflow_description,
-                                workflow_interval,
-                                workflow_disabled,
-                            ) = WorkflowStore.pre_parse_workflow_yaml(workflow_yaml)
+                            pre_parsed_workflow = WorkflowStore.pre_parse_workflow_yaml(
+                                workflow_yaml
+                            )
                         add_or_update_workflow(
-                            id=workflow_id,
-                            name=workflow_name,
+                            id=pre_parsed_workflow.id,
+                            name=pre_parsed_workflow.name,
                             tenant_id=tenant_id,
-                            description=workflow_description,
+                            description=pre_parsed_workflow.description,
                             created_by="system",
                             updated_by="system",
-                            interval=workflow_interval,
-                            is_disabled=workflow_disabled,
+                            interval=pre_parsed_workflow.interval,
+                            is_disabled=pre_parsed_workflow.disabled,
                             workflow_raw=cyaml.dump(workflow_yaml, width=99999),
                             provisioned=True,
                             provisioned_file=workflow_path,
@@ -577,10 +593,12 @@ def get_workflow_meta_data(
                 return providers_dto, triggers
 
             # Parse the workflow YAML safely
-            workflow_yaml = cyaml.safe_load(workflow_raw_data)
-            if not workflow_yaml:
+            workflow_yaml_dict = cyaml.safe_load(workflow_raw_data)
+            if workflow_yaml_dict.get("workflow"):
+                workflow_yaml_dict = workflow_yaml_dict.get("workflow")
+            if not workflow_yaml_dict:
                 self.logger.error(
-                    f"Parsed workflow_yaml is empty or invalid: {workflow_raw_data}"
+                    f"Parsed workflow_yaml is empty or invalid: {workflow_yaml_dict}"
                 )
                 return providers_dto, triggers
 
@@ -595,7 +613,7 @@ def get_workflow_meta_data(
             )  # Return empty providers and triggers in case of error
 
         try:
-            providers = self.parser.get_providers_from_workflow(workflow_yaml)
+            providers = self.parser.get_providers_from_workflow_dict(workflow_yaml_dict)
         except Exception as e:
             self.logger.error(
                 f"Failed to get providerts from workflow: {e}, workflow: {workflow}"
@@ -641,7 +659,7 @@ def get_workflow_meta_data(
                 providers_dto.append(provider_dto)
 
         # Step 3: Extract triggers from workflow
-        triggers = self.parser.get_triggers_from_workflow(workflow_yaml)
+        triggers = self.parser.get_triggers_from_workflow_dict(workflow_yaml_dict)
 
         return providers_dto, triggers
 
diff --git a/tests/test_workflowstore.py b/tests/test_workflowstore.py
index ccf9fd0cd6..41c0b5cac6 100644
--- a/tests/test_workflowstore.py
+++ b/tests/test_workflowstore.py
@@ -6,53 +6,56 @@
 import pytest
 
 VALID_WORKFLOW = """
-id: retrieve-cloudwatch-logs
-name: Retrieve CloudWatch Logs
-description: Retrieve CloudWatch Logs
-triggers:
-  - type: manual
-
-steps:
-  - name: cw-logs
-    provider:
-      config: "{{ providers.cloudwatch }}"
-      type: cloudwatch
-      with:
-        log_groups: 
-         - "meow_logs"
-        query: "fields @message | sort @timestamp desc | limit 20"
-        hours: 4000
-        remove_ptr_from_results: true
+workflow:
+  id: retrieve-cloudwatch-logs
+  name: Retrieve CloudWatch Logs
+  description: Retrieve CloudWatch Logs
+  triggers:
+    - type: manual
+  steps:
+    - name: cw-logs
+      provider:
+        config: "{{ providers.cloudwatch }}"
+        type: cloudwatch
+        with:
+          log_groups: 
+            - "meow_logs"
+          query: "fields @message | sort @timestamp desc | limit 20"
+          hours: 4000
+          remove_ptr_from_results: true
 """
 
 INVALID_WORKFLOW = """
-id: retrieve-cloudwatch-logs
-name: Retrieve CloudWatch Logs
-description: Retrieve CloudWatch Logs
-triggers:
-  - type: manual
-
-steps:
-  - name: cw-logs
-    provider:
-      config: "{{ providers.cloudwatch }}"
-      type: cloudwatch
+workflow:
+  id: retrieve-cloudwatch-logs
+  name: Retrieve CloudWatch Logs
+  description: Retrieve CloudWatch Logs
+  triggers:
+    - type: manual
+  steps:
+    - name: cw-logs
+      provider:
+        config: "{{ providers.cloudwatch }}"
+        type: cloudwatch
+        with:
+          log_groups: 
+            - "meow_logs"
+          query: "fields @message | sort @timestamp desc | limit 20"
+          hours: 4000
+          remove_ptr_from_results: true
+  actions:
+    - name: print-logs
+      if: keep.len({{ steps.cw-logs.results }}) > 0
+      type: print
       with:
-        log_groups: 
-         - "meow_logs"
-        query: "fields @message | sort @timestamp desc | limit 20"
-        hours: 4000
-        remove_ptr_from_results: true
-
-actions:
-  - name: print-logs
-    if: keep.len({{ steps.cw-logs.results }}) > 0
-    type: print
-    with:
-      message: "{{ steps.cw-logs.results }}"
+        message: "{{ steps.cw-logs.results }}"
 """
 
 
+def compare_workflow_raws(a, b):
+    return a.replace(" ", "").replace("\n", "") == b.replace(" ", "").replace("\n", "")
+
+
 def test_get_workflow_meta_data_3832():
     valid_workflow = Workflow(
         id="valid-workflow",
@@ -142,3 +145,88 @@ def test_provision_workflows_no_duplicates(monkeypatch, db_session, test_app):
         assert first_w.name == second_w.name
         assert first_w.workflow_raw == second_w.workflow_raw
         assert first_w.provisioned_file == second_w.provisioned_file
+
+
+@pytest.mark.parametrize(
+    "test_app",
+    [
+        {
+            "AUTH_TYPE": "NOAUTH",
+            "KEEP_WORKFLOWS_DIRECTORY": "./tests/provision/workflows_3",
+        },
+    ],
+    indirect=True,
+)
+def test_unprovision_workflows(monkeypatch, db_session, test_app):
+    """Test that provisioned workflows are deleted when they are no longer provisioned via env or dir."""
+    # First provisioning
+    WorkflowStore.provision_workflows(SINGLE_TENANT_UUID)
+
+    # Get workflows after first provisioning
+    first_provisioned = get_all_provisioned_workflows(SINGLE_TENANT_UUID)
+    assert len(first_provisioned) == 1  # There is 1 workflow in workflows_3 directory
+
+    monkeypatch.delenv("KEEP_WORKFLOWS_DIRECTORY")
+    WorkflowStore.provision_workflows(SINGLE_TENANT_UUID)
+
+    # Get workflows after second provisioning
+    second_provisioned = get_all_provisioned_workflows(SINGLE_TENANT_UUID)
+    assert len(second_provisioned) == 0
+
+
+@pytest.mark.parametrize(
+    "test_app",
+    [
+        {
+            "AUTH_TYPE": "NOAUTH",
+        },
+    ],
+    indirect=True,
+)
+def test_invalid_workflows_dir(monkeypatch, db_session, test_app):
+    """Test exception is rased when invalid dir is passed as KEEP_WORKFLOWS_DIRECTORY."""
+
+    monkeypatch.setenv("KEEP_WORKFLOWS_DIRECTORY", "./tests/provision/workflows_404")
+
+    # First provisioning
+    with pytest.raises(FileNotFoundError):
+        WorkflowStore.provision_workflows(SINGLE_TENANT_UUID)
+
+    # Get workflows after first provisioning
+    provisioned = get_all_provisioned_workflows(SINGLE_TENANT_UUID)
+    assert len(provisioned) == 0  # No workflows has been provisioned
+
+
+@pytest.mark.parametrize(
+    "test_app",
+    [
+        {
+            "AUTH_TYPE": "NOAUTH",
+            "KEEP_WORKFLOWS_DIRECTORY": "./tests/provision/workflows_1",
+        },
+    ],
+    indirect=True,
+)
+def test_change_workflow_provision_method(monkeypatch, db_session, test_app):
+    """Test that provisioned workflows are deleted when they are no longer provisioned via env or dir."""
+    # First provisioning
+    WorkflowStore.provision_workflows(SINGLE_TENANT_UUID)
+
+    # Get workflows after first provisioning
+    first_provisioned = get_all_provisioned_workflows(SINGLE_TENANT_UUID)
+    assert len(first_provisioned) == 3  # There is 3 workflows in workflows_1 directory
+
+    # Provision from env instead of dir
+    monkeypatch.delenv("KEEP_WORKFLOWS_DIRECTORY")
+    monkeypatch.setenv("KEEP_WORKFLOW", VALID_WORKFLOW)
+
+    WorkflowStore.provision_workflows(SINGLE_TENANT_UUID)
+
+    # Get workflows after second provisioning
+    second_provisioned = get_all_provisioned_workflows(SINGLE_TENANT_UUID)
+    assert len(second_provisioned) == 1
+    assert second_provisioned[0].name == "Retrieve CloudWatch Logs"
+    assert (
+        compare_workflow_raws(second_provisioned[0].workflow_raw, VALID_WORKFLOW)
+        == True
+    )


Code review comments:
diff_hunk: @@ -330,43 +340,53 @@ def provision_workflows(
                 f"Directory {provisioned_workflows_dir} does not exist"
             )
 
-        # Get all existing provisioned workflows
-        provisioned_workflows = get_all_provisioned_workflows(tenant_id)
-
         ### Provisioning from env var
         if provisioned_workflow_yaml is not None:
+            pre_parsed_workflow = None
             try:
                 workflow_yaml = cyaml.safe_load(provisioned_workflow_yaml)
-                (
-                    workflow_id,
-                    workflow_name,
-                    workflow_description,
-                    workflow_interval,
-                    workflow_disabled,
-                ) = WorkflowStore.pre_parse_workflow_yaml(workflow_yaml)
+                pre_parsed_workflow = WorkflowStore.pre_parse_workflow_yaml(
+                    workflow_yaml
+                )
+            except ValueError as e:
+                logger.error(
+                    "Error provisioning workflow from env var: yaml is invalid",
+                    extra={"exception": e},
+                )
 
+            try:
                 # Un-provisioning other workflows.
                 for workflow in provisioned_workflows:
-                    if not workflow.id == workflow_id:
-                        logger.info(
-                            f"Deprovisioning workflow {workflow.id} as its id doesn't match the provisioned workflow provided in the env"
-                        )
-                        delete_workflow_by_provisioned_file(
-                            tenant_id, workflow.provisioned_file
-                        )
+                    if (
+                        not pre_parsed_workflow
+                        or not workflow.id == pre_parsed_workflow.id
path: keep/workflowmanager/workflowstore.py
text: @user1:
I believe comparing `id` here is incorrect, as `pre_parsed_workflow.id` (which comes from the provisioning workflow) returns a random UUID. It would be better to compare `name` instead.

@author:
ah, nice catch