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 <abhishek@a6k.me>

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)

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)
 

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 {}),
+        },
+    )