diff --git a/api/db/workflow_run_client.py b/api/db/workflow_run_client.py index c8dc5e34..95e05f3a 100644 --- a/api/db/workflow_run_client.py +++ b/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() diff --git a/api/services/campaign/campaign_call_dispatcher.py b/api/services/campaign/campaign_call_dispatcher.py index bdca8d28..d50a639a 100644 --- a/api/services/campaign/campaign_call_dispatcher.py +++ b/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, diff --git a/api/services/workflow/run_creation.py b/api/services/workflow/run_creation.py index 204496dd..ad5b19ae 100644 --- a/api/services/workflow/run_creation.py +++ b/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 {} diff --git a/api/tests/test_workflow_run_creation.py b/api/tests/test_workflow_run_creation.py index 41f49278..51203fd4 100644 --- a/api/tests/test_workflow_run_creation.py +++ b/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 == {} diff --git a/api/tests/test_workflow_versioning.py b/api/tests/test_workflow_versioning.py index 3739d14f..f89d1890 100644 --- a/api/tests/test_workflow_versioning.py +++ b/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, + )