From 90c5831984ae55ff144e78bc499d15055aea769a Mon Sep 17 00:00:00 2001 From: Sabiha Khan <87858386+chewwbaka@users.noreply.github.com> Date: Tue, 21 Jul 2026 13:24:51 +0530 Subject: [PATCH] fix: move draft and template context handling out of create_workflow_run (#560) * fix: move draft and template context handling out of create_workflow_run create_workflow_run now only creates the run with the definition_id and initial_context provided by the caller. Test call paths explicitly resolve draft definitions and merge template context variables before creating the run, while production/runtime paths bind the published definition without adding template defaults. * fix: review comments * chore: format and minor cleanups --------- Co-authored-by: Abhishek Kumar --- api/db/workflow_run_client.py | 50 ++---- api/routes/agent_stream.py | 4 +- api/routes/public_agent.py | 12 +- api/routes/public_embed.py | 8 + api/routes/telephony.py | 32 ++-- api/routes/workflow.py | 17 +- api/routes/workflow_text_chat.py | 17 +- .../campaign/campaign_call_dispatcher.py | 10 +- api/services/telephony/ari_manager.py | 3 + api/services/workflow/run_creation.py | 57 +++++++ api/tests/test_agent_stream_route.py | 6 +- api/tests/test_from_number_pool_isolation.py | 7 +- api/tests/test_public_agent_routes.py | 151 +++++++++++++++++- api/tests/test_public_embed_cors.py | 12 ++ api/tests/test_telephony_routes.py | 81 ++++++++++ api/tests/test_workflow_create_route.py | 51 +++++- api/tests/test_workflow_run_creation.py | 146 +++++++++++++++++ api/tests/test_workflow_text_chat.py | 12 +- api/tests/test_workflow_versioning.py | 53 ++++-- 19 files changed, 657 insertions(+), 72 deletions(-) create mode 100644 api/services/workflow/run_creation.py create mode 100644 api/tests/test_workflow_run_creation.py diff --git a/api/db/workflow_run_client.py b/api/db/workflow_run_client.py index a9cd7059..a75b8389 100644 --- a/api/db/workflow_run_client.py +++ b/api/db/workflow_run_client.py @@ -33,8 +33,8 @@ class WorkflowRunClient(BaseDBClient): logs: dict = None, campaign_id: int = None, queued_run_id: int = None, - use_draft: bool = False, organization_id: int | None = None, + definition_id: int | None = None, ) -> WorkflowRunModel: async with self.async_session() as session: workflow_query = ( @@ -54,56 +54,28 @@ class WorkflowRunClient(BaseDBClient): if not workflow: raise ValueError(f"Workflow with ID {workflow_id} not found") - # Resolve which definition to bind to this run - target_def = None - - if use_draft: - # For test calls: prefer draft if it exists, fall back to published - draft_result = await session.execute( - select(WorkflowDefinitionModel).where( + if definition_id is not None: + definition_result = await session.execute( + select(WorkflowDefinitionModel.id).where( + WorkflowDefinitionModel.id == definition_id, WorkflowDefinitionModel.workflow_id == workflow.id, - WorkflowDefinitionModel.status == "draft", ) ) - target_def = draft_result.scalars().first() - - if target_def is None: - # Use the published version via released_definition_id (preferred) - # or fall back to is_current for backward compatibility - if workflow.released_definition_id: - target_def = await session.get( - WorkflowDefinitionModel, workflow.released_definition_id + if definition_result.scalar_one_or_none() is None: + raise ValueError( + f"Workflow definition {definition_id} does not belong to " + f"workflow {workflow.id}" ) - else: - pub_result = await session.execute( - select(WorkflowDefinitionModel).where( - WorkflowDefinitionModel.workflow_id == workflow.id, - WorkflowDefinitionModel.is_current == True, - ) - ) - target_def = pub_result.scalars().first() # Get the current storage backend based on ENABLE_AWS_S3 flag current_backend = StorageBackend.get_current_backend() - # Use initial_context from the version if available, else from workflow - default_context = ( - target_def.template_context_variables - if target_def and target_def.template_context_variables - else workflow.template_context_variables - ) - - merged_initial_context = { - **(default_context or {}), - **(initial_context or {}), - } - new_run = WorkflowRunModel( name=name, workflow=workflow, mode=mode, - definition_id=target_def.id if target_def else None, - initial_context=merged_initial_context, + definition_id=definition_id, + initial_context=initial_context or {}, gathered_context=gathered_context or {}, logs=logs or {}, campaign_id=campaign_id, diff --git a/api/routes/agent_stream.py b/api/routes/agent_stream.py index a14ed603..413a5c2b 100644 --- a/api/routes/agent_stream.py +++ b/api/routes/agent_stream.py @@ -24,6 +24,7 @@ from api.services.call_concurrency import ( ) from api.services.quota_service import authorize_workflow_run_start from api.services.telephony import registry as telephony_registry +from api.services.workflow.run_creation import prepare_workflow_run_inputs router = APIRouter(prefix="/agent-stream") @@ -68,11 +69,11 @@ async def agent_stream_websocket( numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 workflow_run_name = f"WR-AGS-{numeric_suffix:08d}" initial_context = { - **(workflow.template_context_variables or {}), "provider": provider_name, "direction": "inbound", } try: + run_inputs = await prepare_workflow_run_inputs(db_client, workflow) workflow_run = await db_client.create_workflow_run( workflow_run_name, workflow.id, @@ -81,6 +82,7 @@ async def agent_stream_websocket( call_type=CallType.INBOUND, initial_context=initial_context, organization_id=workflow.organization_id, + definition_id=run_inputs.definition_id, ) await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) except Exception: diff --git a/api/routes/public_agent.py b/api/routes/public_agent.py index 2b4e95c9..45263534 100644 --- a/api/routes/public_agent.py +++ b/api/routes/public_agent.py @@ -23,6 +23,7 @@ from api.services.telephony.factory import ( get_default_telephony_provider, get_telephony_provider_by_id, ) +from api.services.workflow.run_creation import prepare_workflow_run_inputs from api.utils.common import get_backend_endpoints router = APIRouter(prefix="/public/agent") @@ -260,14 +261,21 @@ async def _execute_resolved_target( ) try: + run_inputs = await prepare_workflow_run_inputs( + db_client, + target.workflow, + initial_context=initial_context, + use_draft=use_draft, + include_template_context=use_draft, + ) workflow_run = await db_client.create_workflow_run( name=workflow_run_name, workflow_id=target.workflow.id, mode=workflow_run_mode, - initial_context=initial_context, + initial_context=run_inputs.initial_context, user_id=execution_user_id, - use_draft=use_draft, organization_id=target.organization_id, + definition_id=run_inputs.definition_id, ) await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) except Exception: diff --git a/api/routes/public_embed.py b/api/routes/public_embed.py index a6126eb5..3ab35f74 100644 --- a/api/routes/public_embed.py +++ b/api/routes/public_embed.py @@ -27,6 +27,7 @@ from api.routes.turn_credentials import ( TurnCredentialsResponse, generate_turn_credentials, ) +from api.services.workflow.run_creation import prepare_workflow_run_inputs router = APIRouter(prefix="/public/embed") @@ -304,6 +305,12 @@ async def initialize_embed_session( # Create workflow run try: + workflow = await db_client.get_workflow( + embed_token.workflow_id, organization_id=embed_token.organization_id + ) + if not workflow: + raise ValueError("Workflow not found") + run_inputs = await prepare_workflow_run_inputs(db_client, workflow) workflow_run = await db_client.create_workflow_run( name=f"Embed Run - {datetime.now(UTC).isoformat()}", workflow_id=embed_token.workflow_id, @@ -314,6 +321,7 @@ async def initialize_embed_session( **(init_request.context_variables or {}), "provider": WorkflowRunMode.SMALLWEBRTC.value, }, + definition_id=run_inputs.definition_id, ) except Exception as e: logger.error(f"Failed to create workflow run: {e}") diff --git a/api/routes/telephony.py b/api/routes/telephony.py index 915d3911..8266f4ab 100644 --- a/api/routes/telephony.py +++ b/api/routes/telephony.py @@ -42,6 +42,7 @@ from api.services.telephony.transfer_event_protocol import ( TransferEvent, TransferEventType, ) +from api.services.workflow.run_creation import prepare_workflow_run_inputs from api.utils.common import get_backend_endpoints from api.utils.telephony_helper import ( generic_hangup_response, @@ -173,27 +174,29 @@ async def initiate_call( try: if not workflow_run_id: - # Merge template context variables (e.g. caller_number, called_number - # set in workflow settings for testing pre-call data fetch). - template_vars = workflow.template_context_variables or {} - numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 workflow_run_name = f"WR-TEL-OUT-{numeric_suffix:08d}" - workflow_run = await db_client.create_workflow_run( - workflow_run_name, - workflow.id, - workflow_run_mode, - user_id=execution_user_id, - call_type=CallType.OUTBOUND, + run_inputs = await prepare_workflow_run_inputs( + db_client, + workflow, initial_context={ - **template_vars, "phone_number": phone_number, "called_number": phone_number, "provider": provider.PROVIDER_NAME, "telephony_configuration_id": telephony_configuration_id, }, use_draft=True, + include_template_context=True, + ) + workflow_run = await db_client.create_workflow_run( + workflow_run_name, + workflow.id, + workflow_run_mode, + user_id=execution_user_id, + call_type=CallType.OUTBOUND, + initial_context=run_inputs.initial_context, organization_id=user.selected_organization_id, + definition_id=run_inputs.definition_id, ) workflow_run_id = workflow_run.id else: @@ -463,6 +466,12 @@ async def _create_inbound_workflow_run( call_id = normalized_data.call_id numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 workflow_run_name = f"WR-TEL-IN-{numeric_suffix:08d}" + workflow = await db_client.get_workflow( + workflow_id, organization_id=organization_id + ) + if not workflow: + raise HTTPException(status_code=404, detail="Workflow not found") + run_inputs = await prepare_workflow_run_inputs(db_client, workflow) workflow_run = await db_client.create_workflow_run( workflow_run_name, @@ -490,6 +499,7 @@ async def _create_inbound_workflow_run( }, }, organization_id=organization_id, + definition_id=run_inputs.definition_id, ) logger.info( diff --git a/api/routes/workflow.py b/api/routes/workflow.py index aefd1441..3e63313f 100644 --- a/api/routes/workflow.py +++ b/api/routes/workflow.py @@ -52,6 +52,7 @@ from api.services.workflow.configuration_policy import ( from api.services.workflow.dto import ReactFlowDTO, sanitize_workflow_definition from api.services.workflow.duplicate import duplicate_workflow from api.services.workflow.errors import ItemKind, WorkflowError +from api.services.workflow.run_creation import prepare_workflow_run_inputs from api.services.workflow.run_usage_response import ( format_public_cost_info, format_public_usage_info, @@ -1317,13 +1318,27 @@ async def create_workflow_run( request: The create workflow run request user: The user to create the workflow run for """ + workflow = await db_client.get_workflow( + workflow_id, organization_id=user.selected_organization_id + ) + if not workflow: + raise HTTPException(status_code=404, detail="Workflow not found") + + run_inputs = await prepare_workflow_run_inputs( + db_client, + workflow, + use_draft=True, + include_template_context=True, + ) + run = await db_client.create_workflow_run( request.name, workflow_id, request.mode, user.id, - use_draft=True, organization_id=user.selected_organization_id, + definition_id=run_inputs.definition_id, + initial_context=run_inputs.initial_context, ) return { "id": run.id, diff --git a/api/routes/workflow_text_chat.py b/api/routes/workflow_text_chat.py index 649cfb71..d6db5529 100644 --- a/api/routes/workflow_text_chat.py +++ b/api/routes/workflow_text_chat.py @@ -11,6 +11,7 @@ from api.db.models import UserModel, WorkflowRunTextSessionModel from api.enums import WorkflowRunMode from api.services.auth.depends import get_user_with_selected_organization from api.services.quota_service import authorize_workflow_run_start +from api.services.workflow.run_creation import prepare_workflow_run_inputs from api.services.workflow.text_chat_session_service import ( TextChatPendingTurnLostError, TextChatSessionExecutionError, @@ -164,14 +165,26 @@ async def create_text_chat_session( ) -> WorkflowRunTextSessionResponse: session_name = request.name or f"WR-TEXT-{uuid4().hex[:6].upper()}" try: + workflow = await db_client.get_workflow( + workflow_id, organization_id=user.selected_organization_id + ) + if not workflow: + raise ValueError(f"Workflow with ID {workflow_id} not found") + run_inputs = await prepare_workflow_run_inputs( + db_client, + workflow, + initial_context=request.initial_context, + use_draft=True, + include_template_context=True, + ) workflow_run = await db_client.create_workflow_run( name=session_name, workflow_id=workflow_id, mode=WorkflowRunMode.TEXTCHAT.value, user_id=user.id, - initial_context=request.initial_context, - use_draft=True, + initial_context=run_inputs.initial_context, organization_id=user.selected_organization_id, + definition_id=run_inputs.definition_id, ) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) diff --git a/api/services/campaign/campaign_call_dispatcher.py b/api/services/campaign/campaign_call_dispatcher.py index 61411f8d..39688473 100644 --- a/api/services/campaign/campaign_call_dispatcher.py +++ b/api/services/campaign/campaign_call_dispatcher.py @@ -20,6 +20,7 @@ from api.services.campaign.errors import ( ) from api.services.campaign.rate_limiter import rate_limiter from api.services.quota_service import authorize_workflow_run_start +from api.services.workflow.run_creation import prepare_workflow_run_inputs from api.utils.common import get_backend_endpoints if TYPE_CHECKING: @@ -237,7 +238,10 @@ class CampaignCallDispatcher: try: # Get workflow details - workflow = await db_client.get_workflow_by_id(campaign.workflow_id) + workflow = await db_client.get_workflow( + campaign.workflow_id, + organization_id=campaign.organization_id, + ) if not workflow: raise ValueError(f"Workflow {campaign.workflow_id} not found") @@ -280,6 +284,7 @@ class CampaignCallDispatcher: # Create workflow run with queued_run_id tracking workflow_run_name = f"WR-CAMPAIGN-{campaign.id}-{queued_run.id}" + run_inputs = await prepare_workflow_run_inputs(db_client, workflow) workflow_run = await db_client.create_workflow_run( name=workflow_run_name, workflow_id=campaign.workflow_id, @@ -289,6 +294,7 @@ class CampaignCallDispatcher: campaign_id=campaign.id, queued_run_id=queued_run.id, # Link to queued run for retry tracking organization_id=campaign.organization_id, + definition_id=run_inputs.definition_id, ) await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) slot_bound = True @@ -300,7 +306,7 @@ class CampaignCallDispatcher: from_number, telephony_configuration_id=campaign.telephony_configuration_id, ) - except Exception as e: + except Exception: # Release slot and from_number on error if slot_bound and workflow_run: await call_concurrency.release_workflow_run_slot(workflow_run.id) diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index 1c0a805d..6f5d3b9a 100644 --- a/api/services/telephony/ari_manager.py +++ b/api/services/telephony/ari_manager.py @@ -38,6 +38,7 @@ from api.services.telephony.transfer_event_protocol import ( TransferEvent, TransferEventType, ) +from api.services.workflow.run_creation import prepare_workflow_run_inputs # Redis key pattern and TTL for channel-to-run mapping _CHANNEL_KEY_PREFIX = "ari:channel:" @@ -633,6 +634,7 @@ class ARIConnection: # 3. Create workflow run call_id = channel_id + run_inputs = await prepare_workflow_run_inputs(db_client, workflow) # Capture the configured external PBX identity from SIP headers. external_pbx_call = await self._capture_external_pbx_call( channel_id, channel.get("name", "") @@ -655,6 +657,7 @@ class ARIConnection: "call_id": call_id, }, organization_id=self.organization_id, + definition_id=run_inputs.definition_id, ) await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) diff --git a/api/services/workflow/run_creation.py b/api/services/workflow/run_creation.py new file mode 100644 index 00000000..ad5b19ae --- /dev/null +++ b/api/services/workflow/run_creation.py @@ -0,0 +1,57 @@ +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class WorkflowRunInputs: + definition_id: int | None + initial_context: dict[str, Any] + + +def _published_definition(workflow) -> object | None: + return getattr(workflow, "released_definition", None) or getattr( + workflow, "current_definition", None + ) + + +async def prepare_workflow_run_inputs( + workflow_client, + workflow, + *, + initial_context: dict[str, Any] | None = None, + use_draft: bool = False, + include_template_context: bool = False, +) -> WorkflowRunInputs: + """Resolve definition binding and optional template defaults for a run. + + Draft and template-context handling belong at runtime call sites, not in the + persistence client. Callers must opt in explicitly for workflow-editor/test + flows. + """ + target_definition = None + if use_draft: + target_definition = await workflow_client.get_draft_version(workflow.id) + + if target_definition is None: + target_definition = _published_definition(workflow) + + default_context = {} + if include_template_context: + definition_context = ( + getattr(target_definition, "template_context_variables", None) + if target_definition + else None + ) + default_context = ( + definition_context + if definition_context is not None + else getattr(workflow, "template_context_variables", None) + ) or {} + + return WorkflowRunInputs( + definition_id=getattr(target_definition, "id", None), + initial_context={ + **default_context, + **(initial_context or {}), + }, + ) diff --git a/api/tests/test_agent_stream_route.py b/api/tests/test_agent_stream_route.py index 5a143c07..1ddf6d3f 100644 --- a/api/tests/test_agent_stream_route.py +++ b/api/tests/test_agent_stream_route.py @@ -28,6 +28,8 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): user_id=22, organization_id=33, template_context_variables={"existing": "context"}, + released_definition=SimpleNamespace(id=55, template_context_variables={}), + current_definition=None, ) workflow_run = SimpleNamespace(id=44) provider = SimpleNamespace(handle_external_websocket=AsyncMock()) @@ -72,10 +74,10 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): assert create_args[2] == "cloudonix" assert create_kwargs["organization_id"] == workflow.organization_id assert create_kwargs["initial_context"] == { - "existing": "context", "provider": "cloudonix", "direction": "inbound", } + assert create_kwargs["definition_id"] == 55 provider.handle_external_websocket.assert_awaited_once() _, provider_kwargs = provider.handle_external_websocket.await_args assert provider_kwargs["params"] == {"custom": "value"} @@ -92,6 +94,8 @@ async def test_agent_stream_rejects_when_concurrency_limit_reached(): user_id=22, organization_id=33, template_context_variables={}, + released_definition=SimpleNamespace(id=55, template_context_variables={}), + current_definition=None, ) spec = SimpleNamespace(provider_cls=lambda _config: object()) diff --git a/api/tests/test_from_number_pool_isolation.py b/api/tests/test_from_number_pool_isolation.py index 5e393111..4ce498ee 100644 --- a/api/tests/test_from_number_pool_isolation.py +++ b/api/tests/test_from_number_pool_isolation.py @@ -281,7 +281,7 @@ class TestDispatcherThreadsTelephonyConfig: ), ), ): - mock_db.get_workflow_by_id = AsyncMock(return_value=SimpleNamespace(id=1)) + mock_db.get_workflow = AsyncMock(return_value=SimpleNamespace(id=1)) mock_db.create_workflow_run = AsyncMock(return_value=workflow_run) mock_db.update_workflow_run = AsyncMock() mock_concurrency.bind_workflow_run = AsyncMock() @@ -300,6 +300,11 @@ class TestDispatcherThreadsTelephonyConfig: ) await dispatcher.dispatch_call(queued_run, campaign, slot) + mock_db.get_workflow.assert_awaited_once_with( + campaign.workflow_id, + organization_id=org_id, + ) + # acquire_from_number on rate_limiter must be called with the # campaign's telephony_configuration_id. assert mock_rl.acquire_from_number.await_count == 1 diff --git a/api/tests/test_public_agent_routes.py b/api/tests/test_public_agent_routes.py index 3a00ef57..241f9865 100644 --- a/api/tests/test_public_agent_routes.py +++ b/api/tests/test_public_agent_routes.py @@ -31,8 +31,12 @@ def _active_workflow(*, trigger_path: str | None = None): status="active", workflow_uuid="workflow-uuid-123", released_definition=SimpleNamespace( - workflow_json={"nodes": nodes, "edges": []} + id=77, + workflow_json={"nodes": nodes, "edges": []}, + template_context_variables={"name": "published"}, ), + current_definition=None, + template_context_variables={"name": "workflow"}, ) @@ -123,6 +127,9 @@ def test_trigger_route_executes_as_workflow_owner(): assert create_kwargs["initial_context"]["workflow_uuid"] == workflow.workflow_uuid assert create_kwargs["initial_context"]["api_key_id"] == 7 assert create_kwargs["initial_context"]["api_key_created_by"] == 22 + assert create_kwargs["definition_id"] == 77 + assert "name" not in create_kwargs["initial_context"] + assert not mock_db.get_draft_version.called initiate_kwargs = provider.initiate_call.await_args.kwargs assert initiate_kwargs["workflow_id"] == workflow.id @@ -197,6 +204,148 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution(): ) assert create_kwargs["initial_context"]["agent_identifier_type"] == "workflow_uuid" assert "agent_uuid" not in create_kwargs["initial_context"] + assert create_kwargs["definition_id"] == 77 + assert "name" not in create_kwargs["initial_context"] + assert not mock_db.get_draft_version.called + + +def test_trigger_test_route_uses_draft_and_template_context_with_api_override(): + app = _make_test_app() + client = TestClient(app) + + workflow = _active_workflow(trigger_path="trigger-uuid-123") + draft = SimpleNamespace( + id=88, + workflow_json=workflow.released_definition.workflow_json, + template_context_variables={"name": "john", "age": 12, "rank": 2}, + ) + provider = _provider() + quota_mock = AsyncMock( + return_value=SimpleNamespace(has_quota=True, error_message="") + ) + + with ( + patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, + patch( + "api.routes.public_agent.authorize_workflow_run_start", + new=quota_mock, + ), + patch( + "api.routes.public_agent.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + patch( + "api.routes.public_agent.get_backend_endpoints", + new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), + ), + ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + + mock_db.validate_api_key = AsyncMock( + return_value=SimpleNamespace(id=7, organization_id=11, created_by=22) + ) + mock_db.get_agent_trigger_by_path = AsyncMock( + return_value=SimpleNamespace( + workflow_id=workflow.id, + organization_id=11, + state="active", + ) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=draft) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501)) + + response = client.post( + "/public/agent/test/trigger-uuid-123", + headers={"X-API-Key": "test-api-key"}, + json={ + "phone_number": "+15551234567", + "initial_context": {"name": "tom", "age": 10}, + }, + ) + + assert response.status_code == 200 + assert mock_db.get_draft_version.await_count == 2 + create_kwargs = mock_db.create_workflow_run.await_args.kwargs + assert create_kwargs["definition_id"] == draft.id + assert create_kwargs["initial_context"]["name"] == "tom" + assert create_kwargs["initial_context"]["age"] == 10 + assert create_kwargs["initial_context"]["rank"] == 2 + assert create_kwargs["initial_context"]["trigger_mode"] == "test" + + +def test_workflow_uuid_test_route_uses_draft_and_template_context(): + app = _make_test_app() + client = TestClient(app) + + workflow = _active_workflow() + draft = SimpleNamespace( + id=88, + workflow_json={"nodes": [], "edges": []}, + template_context_variables={"name": "john", "age": 12, "rank": 2}, + ) + provider = _provider() + quota_mock = AsyncMock( + return_value=SimpleNamespace(has_quota=True, error_message="") + ) + + with ( + patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, + patch( + "api.routes.public_agent.authorize_workflow_run_start", + new=quota_mock, + ), + patch( + "api.routes.public_agent.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + patch( + "api.routes.public_agent.get_backend_endpoints", + new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), + ), + ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + + mock_db.validate_api_key = AsyncMock( + return_value=SimpleNamespace(id=7, organization_id=11, created_by=22) + ) + mock_db.get_workflow_by_uuid = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=draft) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501)) + + response = client.post( + f"/public/agent/test/workflow/{workflow.workflow_uuid}", + headers={"X-API-Key": "test-api-key"}, + json={ + "phone_number": "+15551234567", + "initial_context": {"name": "tom"}, + }, + ) + + assert response.status_code == 200 + assert mock_db.get_draft_version.await_count == 2 + create_kwargs = mock_db.create_workflow_run.await_args.kwargs + assert create_kwargs["definition_id"] == draft.id + assert create_kwargs["initial_context"]["name"] == "tom" + assert create_kwargs["initial_context"]["age"] == 12 + assert create_kwargs["initial_context"]["rank"] == 2 + assert create_kwargs["initial_context"]["trigger_mode"] == "test" def test_trigger_route_rejects_when_concurrency_limit_reached(): diff --git a/api/tests/test_public_embed_cors.py b/api/tests/test_public_embed_cors.py index 1608fc72..80bb78c6 100644 --- a/api/tests/test_public_embed_cors.py +++ b/api/tests/test_public_embed_cors.py @@ -89,6 +89,14 @@ def _patch_db(monkeypatch): async def _create_workflow_run(**_kwargs): return SimpleNamespace(id=123) + async def _get_workflow(*_args, **_kwargs): + return SimpleNamespace( + id=1, + released_definition=SimpleNamespace(id=55, template_context_variables={}), + current_definition=None, + template_context_variables={}, + ) + async def _noop(*_args, **_kwargs): return None @@ -108,6 +116,10 @@ def _patch_db(monkeypatch): "api.routes.public_embed.db_client.create_workflow_run", _create_workflow_run, ) + monkeypatch.setattr( + "api.routes.public_embed.db_client.get_workflow", + _get_workflow, + ) monkeypatch.setattr( "api.routes.public_embed.db_client.create_embed_session", _noop, diff --git a/api/tests/test_telephony_routes.py b/api/tests/test_telephony_routes.py index b244c968..87673910 100644 --- a/api/tests/test_telephony_routes.py +++ b/api/tests/test_telephony_routes.py @@ -28,6 +28,11 @@ def _workflow(*, workflow_id: int = 33, user_id: int = 99): user_id=user_id, organization_id=11, template_context_variables={"template_key": "template-value"}, + released_definition=SimpleNamespace( + id=77, + template_context_variables={"template_key": "template-value"}, + ), + current_definition=None, ) @@ -84,6 +89,7 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): return_value=SimpleNamespace(id=55) ) mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=None) mock_db.create_workflow_run = AsyncMock( return_value=SimpleNamespace( id=501, @@ -113,6 +119,7 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): assert create_args[1] == workflow.id assert create_kwargs["user_id"] == workflow.user_id assert create_kwargs["organization_id"] == workflow.organization_id + assert create_kwargs["definition_id"] == 77 assert create_kwargs["initial_context"]["template_key"] == "template-value" mock_concurrency.acquire_org_slot.assert_awaited_once_with( workflow.organization_id, @@ -132,6 +139,79 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): mock_db.get_user_configurations.assert_not_called() +def test_initiate_call_uses_draft_template_context_with_provider_overrides(): + app = _make_test_app() + client = TestClient(app) + + workflow = _workflow() + draft = SimpleNamespace( + id=88, + template_context_variables={ + "phone_number": "template-phone", + "called_number": "template-called", + "provider": "template-provider", + "draft_only": "kept", + }, + ) + provider = _provider() + quota_mock = AsyncMock( + return_value=SimpleNamespace(has_quota=True, error_message="") + ) + + with ( + patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, + patch( + "api.routes.telephony.authorize_workflow_run_start", + new=quota_mock, + ), + patch( + "api.routes.telephony.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + patch( + "api.routes.telephony.get_backend_endpoints", + new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), + ), + ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + + mock_db.get_user_configurations = AsyncMock( + return_value=SimpleNamespace(test_phone_number=None) + ) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=draft) + mock_db.create_workflow_run = AsyncMock( + return_value=SimpleNamespace( + id=501, + name="WR-TEL-OUT-00000001", + initial_context={}, + ) + ) + mock_db.update_workflow_run = AsyncMock() + + response = client.post( + "/telephony/initiate-call", + json={"workflow_id": workflow.id, "phone_number": "+15551234567"}, + ) + + assert response.status_code == 200 + mock_db.get_draft_version.assert_awaited_once_with(workflow.id) + create_kwargs = mock_db.create_workflow_run.await_args.kwargs + assert create_kwargs["definition_id"] == draft.id + assert create_kwargs["initial_context"]["phone_number"] == "+15551234567" + assert create_kwargs["initial_context"]["called_number"] == "+15551234567" + assert create_kwargs["initial_context"]["provider"] == provider.PROVIDER_NAME + assert create_kwargs["initial_context"]["draft_only"] == "kept" + + def test_initiate_call_uses_organization_preference_phone_number(): app = _make_test_app() client = TestClient(app) @@ -173,6 +253,7 @@ def test_initiate_call_uses_organization_preference_phone_number(): return_value=SimpleNamespace(id=55) ) mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=None) mock_db.create_workflow_run = AsyncMock( return_value=SimpleNamespace( id=501, diff --git a/api/tests/test_workflow_create_route.py b/api/tests/test_workflow_create_route.py index 6c2b9850..cb1ec579 100644 --- a/api/tests/test_workflow_create_route.py +++ b/api/tests/test_workflow_create_route.py @@ -1,5 +1,6 @@ +from datetime import UTC, datetime from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, patch from fastapi import FastAPI from fastapi.testclient import TestClient @@ -82,3 +83,51 @@ def test_create_workflow_rejects_duplicate_api_triggers_before_db_write(): assert detail["errors"][0]["kind"] == "workflow" assert "at most one API Trigger" in detail["errors"][0]["message"] assert mock_db.mock_calls == [] + + +def test_create_workflow_run_uses_draft_and_template_context(): + app = _make_test_app() + client = TestClient(app) + + workflow = SimpleNamespace( + id=33, + released_definition=SimpleNamespace( + id=77, + template_context_variables={"name": "published"}, + ), + current_definition=None, + template_context_variables={"name": "workflow"}, + ) + draft = SimpleNamespace( + id=88, + template_context_variables={"name": "draft", "draft_only": "kept"}, + ) + run = SimpleNamespace( + id=501, + workflow_id=workflow.id, + name="WR-test", + mode="smallwebrtc", + created_at=datetime.now(UTC), + definition_id=draft.id, + initial_context={"name": "draft", "draft_only": "kept"}, + gathered_context={}, + ) + + with patch("api.routes.workflow.db_client") as mock_db: + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_draft_version = AsyncMock(return_value=draft) + mock_db.create_workflow_run = AsyncMock(return_value=run) + + response = client.post( + f"/workflow/{workflow.id}/runs", + json={"name": "WR-test", "mode": "smallwebrtc"}, + ) + + assert response.status_code == 200 + mock_db.get_draft_version.assert_awaited_once_with(workflow.id) + create_kwargs = mock_db.create_workflow_run.await_args.kwargs + assert create_kwargs["definition_id"] == draft.id + assert create_kwargs["initial_context"] == { + "name": "draft", + "draft_only": "kept", + } diff --git a/api/tests/test_workflow_run_creation.py b/api/tests/test_workflow_run_creation.py new file mode 100644 index 00000000..51203fd4 --- /dev/null +++ b/api/tests/test_workflow_run_creation.py @@ -0,0 +1,146 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from api.services.workflow.run_creation import prepare_workflow_run_inputs + + +def _workflow(): + return SimpleNamespace( + id=33, + template_context_variables={"name": "workflow", "workflow_only": "kept"}, + released_definition=SimpleNamespace( + id=77, + template_context_variables={ + "name": "published", + "published_only": "kept", + }, + ), + current_definition=None, + ) + + +@pytest.mark.asyncio +async def test_prepare_inputs_uses_draft_and_merges_template_context(): + draft = SimpleNamespace( + id=88, + template_context_variables={"name": "draft", "draft_only": "kept"}, + ) + workflow_client = SimpleNamespace(get_draft_version=AsyncMock(return_value=draft)) + + run_inputs = await prepare_workflow_run_inputs( + workflow_client, + _workflow(), + initial_context={"name": "explicit", "explicit_only": "kept"}, + use_draft=True, + include_template_context=True, + ) + + assert run_inputs.definition_id == 88 + assert run_inputs.initial_context == { + "name": "explicit", + "draft_only": "kept", + "explicit_only": "kept", + } + workflow_client.get_draft_version.assert_awaited_once_with(33) + + +@pytest.mark.asyncio +async def test_prepare_inputs_does_not_check_draft_or_merge_templates_by_default(): + workflow_client = SimpleNamespace(get_draft_version=AsyncMock()) + + run_inputs = await prepare_workflow_run_inputs( + workflow_client, + _workflow(), + initial_context={"name": "explicit"}, + ) + + assert run_inputs.definition_id == 77 + assert run_inputs.initial_context == {"name": "explicit"} + workflow_client.get_draft_version.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_prepare_inputs_falls_back_to_published_when_draft_missing(): + workflow_client = SimpleNamespace(get_draft_version=AsyncMock(return_value=None)) + + run_inputs = await prepare_workflow_run_inputs( + workflow_client, + _workflow(), + initial_context={"name": "explicit"}, + use_draft=True, + include_template_context=True, + ) + + assert run_inputs.definition_id == 77 + assert run_inputs.initial_context == { + "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_text_chat.py b/api/tests/test_workflow_text_chat.py index 7b964935..4929fae9 100644 --- a/api/tests/test_workflow_text_chat.py +++ b/api/tests/test_workflow_text_chat.py @@ -198,6 +198,11 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn( workflow_definition=workflow_definition, suffix="bootstrap", ) + draft = await db_session.save_workflow_draft( + workflow_id=workflow.id, + workflow_definition=workflow_definition, + template_context_variables={"name": "draft", "draft_only": "kept"}, + ) llm = MockLLMService( mock_steps=[ @@ -219,7 +224,7 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn( ): create_response = await client.post( f"/api/v1/workflow/{workflow.id}/text-chat/sessions", - json={}, + json={"initial_context": {"name": "explicit"}}, ) assert create_response.status_code == 200 created = create_response.json() @@ -242,6 +247,11 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn( assert "Start" in (created["gathered_context"] or {}).get("nodes_visited", []) workflow_run = await db_session.get_workflow_run_by_id(created["workflow_run_id"]) assert workflow_run is not None + assert workflow_run.definition_id == draft.id + assert workflow_run.initial_context == { + "name": "explicit", + "draft_only": "kept", + } assert "call_duration_seconds" in workflow_run.usage_info assert _log_texts(run_payload["logs"], "rtf-bot-text") == [ "Hello from the workflow tester." diff --git a/api/tests/test_workflow_versioning.py b/api/tests/test_workflow_versioning.py index c27d0f88..f89d1890 100644 --- a/api/tests/test_workflow_versioning.py +++ b/api/tests/test_workflow_versioning.py @@ -574,21 +574,25 @@ class TestRunDefinitionBinding: workflow_definition=GRAPH_V2, ) + versions = await db_session.get_workflow_versions(workflow.id) + published = next(v for v in versions if v.status == "published") + # Create a run (simulating campaign dispatch) run = await db_session.create_workflow_run( name="Campaign Run", workflow_id=workflow.id, mode="webrtc", user_id=user.id, + definition_id=published.id, ) - # Run should be bound to the published V1, not the draft V2 - versions = await db_session.get_workflow_versions(workflow.id) - published = next(v for v in versions if v.status == "published") + # Run should be bound to the caller-selected published V1. assert run.definition_id == published.id - async def test_test_run_uses_draft_if_exists(self, db_session, workflow_with_v1): - """A test/phone call should use the draft version for pre-publish testing.""" + async def test_run_uses_caller_supplied_draft_definition( + self, db_session, workflow_with_v1 + ): + """Test/phone callers bind drafts before creating the run.""" workflow, user = workflow_with_v1 draft = await db_session.save_workflow_draft( @@ -602,15 +606,15 @@ class TestRunDefinitionBinding: workflow_id=workflow.id, mode="webrtc", # test mode user_id=user.id, - use_draft=True, + definition_id=draft.id, ) assert run.definition_id == draft.id - async def test_run_initial_context_merges_with_template_context( + async def test_run_initial_context_is_stored_as_provided( self, db_session, workflow_with_v1 ): - """Explicit run context should augment template context, not replace it.""" + """Template context merging is handled before calling the DB client.""" workflow, user = workflow_with_v1 await db_session.save_workflow_draft( workflow_id=workflow.id, @@ -634,6 +638,37 @@ class TestRunDefinitionBinding: assert run.initial_context == { "company_name": "Override Co", - "default_only": "kept", "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, + )