fix: review comments

api/db/workflow_run_client.py

@@ -10,6 +10,7 @@ from api.db.filters import apply_workflow_run_filters, get_workflow_run_order_cl
 from api.db.models import (
     OrganizationModel,
     UserModel,
+    WorkflowDefinitionModel,
     WorkflowModel,
     WorkflowRunModel,
 )
@@ -53,6 +54,14 @@ class WorkflowRunClient(BaseDBClient):
             if not workflow:
                 raise ValueError(f"Workflow with ID {workflow_id} not found")
 
+            if definition_id is not None:
+                definition = await session.get(WorkflowDefinitionModel, definition_id)
+                if not definition or definition.workflow_id != workflow.id:
+                    raise ValueError(
+                        f"Workflow definition {definition_id} does not belong to "
+                        f"workflow {workflow.id}"
+                    )
+
             # Get the current storage backend based on ENABLE_AWS_S3 flag
             current_backend = StorageBackend.get_current_backend()
 

api/services/campaign/campaign_call_dispatcher.py

@@ -241,6 +241,11 @@ class CampaignCallDispatcher:
             workflow = await db_client.get_workflow_by_id(campaign.workflow_id)
             if not workflow:
                 raise ValueError(f"Workflow {campaign.workflow_id} not found")
+            if workflow.organization_id != campaign.organization_id:
+                raise ValueError(
+                    f"Workflow {campaign.workflow_id} does not belong to "
+                    f"organization {campaign.organization_id}"
+                )
 
             # Extract phone number
             phone_number = queued_run.context_variables.get("phone_number")
@@ -281,11 +286,6 @@ class CampaignCallDispatcher:
 
             # Create workflow run with queued_run_id tracking
             workflow_run_name = f"WR-CAMPAIGN-{campaign.id}-{queued_run.id}"
-            workflow = await db_client.get_workflow(
-                campaign.workflow_id, organization_id=campaign.organization_id
-            )
-            if not workflow:
-                raise ValueError(f"Workflow with ID {campaign.workflow_id} not found")
             run_inputs = await prepare_workflow_run_inputs(db_client, workflow)
             workflow_run = await db_client.create_workflow_run(
                 name=workflow_run_name,

api/services/workflow/run_creation.py

@@ -37,10 +37,14 @@ async def prepare_workflow_run_inputs(
 
     default_context = {}
     if include_template_context:
-        default_context = (
-            target_definition.template_context_variables
+        definition_context = (
+            getattr(target_definition, "template_context_variables", None)
             if target_definition
-            and getattr(target_definition, "template_context_variables", None)
+            else None
+        )
+        default_context = (
+            definition_context
+            if definition_context is not None
             else getattr(workflow, "template_context_variables", None)
         ) or {}
 

api/tests/test_workflow_run_creation.py

@@ -78,3 +78,69 @@ async def test_prepare_inputs_falls_back_to_published_when_draft_missing():
         "name": "explicit",
         "published_only": "kept",
     }
+    workflow_client.get_draft_version.assert_awaited_once_with(33)
+
+
+@pytest.mark.asyncio
+async def test_prepare_inputs_uses_current_definition_when_released_missing():
+    workflow = _workflow()
+    workflow.released_definition = None
+    workflow.current_definition = SimpleNamespace(
+        id=66,
+        template_context_variables={"name": "current", "current_only": "kept"},
+    )
+    workflow_client = SimpleNamespace(get_draft_version=AsyncMock())
+
+    run_inputs = await prepare_workflow_run_inputs(
+        workflow_client,
+        workflow,
+        include_template_context=True,
+    )
+
+    assert run_inputs.definition_id == 66
+    assert run_inputs.initial_context == {
+        "name": "current",
+        "current_only": "kept",
+    }
+    workflow_client.get_draft_version.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_prepare_inputs_falls_back_to_workflow_template_context_when_definition_context_is_none():
+    workflow = _workflow()
+    workflow.released_definition = SimpleNamespace(
+        id=77,
+        template_context_variables=None,
+    )
+    workflow_client = SimpleNamespace(get_draft_version=AsyncMock())
+
+    run_inputs = await prepare_workflow_run_inputs(
+        workflow_client,
+        workflow,
+        include_template_context=True,
+    )
+
+    assert run_inputs.definition_id == 77
+    assert run_inputs.initial_context == {
+        "name": "workflow",
+        "workflow_only": "kept",
+    }
+
+
+@pytest.mark.asyncio
+async def test_prepare_inputs_respects_empty_definition_template_context():
+    workflow = _workflow()
+    workflow.released_definition = SimpleNamespace(
+        id=77,
+        template_context_variables={},
+    )
+    workflow_client = SimpleNamespace(get_draft_version=AsyncMock())
+
+    run_inputs = await prepare_workflow_run_inputs(
+        workflow_client,
+        workflow,
+        include_template_context=True,
+    )
+
+    assert run_inputs.definition_id == 77
+    assert run_inputs.initial_context == {}

api/tests/test_workflow_versioning.py

@@ -640,3 +640,35 @@ class TestRunDefinitionBinding:
             "company_name": "Override Co",
             "provider": "smallwebrtc",
         }
+
+    async def test_run_rejects_definition_from_another_workflow(
+        self, db_session, org_and_user
+    ):
+        """The DB client must not bind a run to another workflow's definition."""
+        org, user = org_and_user
+        workflow_a = await db_session.create_workflow(
+            name="Workflow A",
+            workflow_definition=GRAPH_V1,
+            user_id=user.id,
+            organization_id=org.id,
+        )
+        workflow_b = await db_session.create_workflow(
+            name="Workflow B",
+            workflow_definition=GRAPH_V2,
+            user_id=user.id,
+            organization_id=org.id,
+        )
+        workflow_b_versions = await db_session.get_workflow_versions(workflow_b.id)
+        workflow_b_definition = next(
+            v for v in workflow_b_versions if v.status == "published"
+        )
+
+        with pytest.raises(ValueError, match="does not belong"):
+            await db_session.create_workflow_run(
+                name="Bad Run",
+                workflow_id=workflow_a.id,
+                mode="smallwebrtc",
+                user_id=user.id,
+                organization_id=org.id,
+                definition_id=workflow_b_definition.id,
+            )